Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
125 changes: 85 additions & 40 deletions contrib/istio/filters/http/istio_stats/source/istio_stats.cc
Original file line number Diff line number Diff line change
Expand Up @@ -114,6 +114,10 @@ enum class Reporter {
ServerSidecar,
// Gateway listener for a set of destination workloads.
ServerGateway,
// The configuration does not specify the
// reporter type and the traffic direction
// will be used to infer the reporter type.
Unspecified,
};

// Detect if peer info read is completed by TCP metadata exchange.
Expand Down Expand Up @@ -474,30 +478,22 @@ struct MetricOverrides : public Logger::Loggable<Logger::Id::filter> {

struct Config : public Logger::Loggable<Logger::Id::filter> {
Config(const stats::PluginConfig& proto_config,
Server::Configuration::FactoryContext& factory_context)
: context_(factory_context.serverFactoryContext().singletonManager().getTyped<Context>(
Server::Configuration::ServerFactoryContext& context, Stats::Scope& stats_scope)
: context_(context.singletonManager().getTyped<Context>(
SINGLETON_MANAGER_REGISTERED_NAME(Context),
[&factory_context] {
return std::make_shared<Context>(factory_context.serverFactoryContext().scope(),
factory_context.serverFactoryContext().localInfo());
[&context] {
return std::make_shared<Context>(context.scope(), context.localInfo());
})),
disable_host_header_fallback_(proto_config.disable_host_header_fallback()),
report_duration_(
PROTOBUF_GET_MS_OR_DEFAULT(proto_config, tcp_reporting_duration, /* 5s */ 5000)) {
recordVersion(factory_context);
recordVersion(stats_scope);
reporter_ = Reporter::ClientSidecar;
switch (proto_config.reporter()) {
case stats::Reporter::UNSPECIFIED:
switch (factory_context.direction()) {
case envoy::config::core::v3::TrafficDirection::INBOUND:
reporter_ = Reporter::ServerSidecar;
break;
case envoy::config::core::v3::TrafficDirection::OUTBOUND:
reporter_ = Reporter::ClientSidecar;
break;
default:
break;
}
// Mark as unspecified to allow the filter to infer the reporter type based on traffic
// direction.
reporter_ = Reporter::Unspecified;
break;
case stats::Reporter::SERVER_GATEWAY:
reporter_ = Reporter::ServerGateway;
Expand Down Expand Up @@ -743,14 +739,13 @@ struct Config : public Logger::Loggable<Logger::Id::filter> {
bool evaluated_{false};
};

void recordVersion(Server::Configuration::FactoryContext& factory_context) {
void recordVersion(Stats::Scope& scope) {
Stats::StatNameTagVector tags;
tags.push_back({context_->component_, context_->proxy_});
tags.push_back({context_->tag_, context_->istio_version_.empty() ? context_->unknown_
: context_->istio_version_});

Stats::Utility::gaugeFromStatNames(factory_context.scope(),
{context_->stat_namespace_, context_->istio_build_},
Stats::Utility::gaugeFromStatNames(scope, {context_->stat_namespace_, context_->istio_build_},
Stats::Gauge::ImportMode::Accumulate, tags)
.set(1);
}
Expand All @@ -776,9 +771,46 @@ class IstioStatsFilter : public Http::PassThroughFilter,
public:
IstioStatsFilter(ConfigSharedPtr config)
: config_(config), context_(*config->context_), pool_(config->scope().symbolTable()),
stream_(*config_, pool_) {
stream_(*config_, pool_) {}
~IstioStatsFilter() override { ASSERT(report_timer_ == nullptr); }

// The reporter type is only set explicitly by gateways/waypoints. Sidecars leave it
// unspecified in the config and it is inferred here from the direction of the listener
// that owns this stream or connection.
Reporter resolveReporter() const {
ASSERT(decoder_callbacks_ != nullptr || network_read_callbacks_ != nullptr);
if (config_->reporter() != Reporter::Unspecified) {
return config_->reporter();
}

OptRef<const Network::ListenerInfo> listener_info;
if (decoder_callbacks_ != nullptr) {
listener_info = decoder_callbacks_->streamInfo().downstreamAddressProvider().listenerInfo();
} else if (network_read_callbacks_ != nullptr) {
listener_info = network_read_callbacks_->connection().connectionInfoProvider().listenerInfo();
}
if (listener_info.has_value()) {
switch (listener_info->direction()) {
case envoy::config::core::v3::TrafficDirection::INBOUND:
return Reporter::ServerSidecar;
case envoy::config::core::v3::TrafficDirection::OUTBOUND:
return Reporter::ClientSidecar;
default:
break;
}
}
// Retain the historical default for listeners without a traffic direction.
return Reporter::ClientSidecar;
}

// Resolved reporter type. Only valid once initializeTags() has run, which is the first
// thing both the HTTP and the network entry point do.
Reporter reporter() const { return reporter_; }

void initializeTags() {
reporter_ = resolveReporter();
tags_.reserve(25);
switch (config_->reporter()) {
switch (reporter_) {
case Reporter::ServerSidecar:
tags_.push_back({context_.reporter_, context_.destination_});
break;
Expand All @@ -788,11 +820,19 @@ class IstioStatsFilter : public Http::PassThroughFilter,
case Reporter::ClientSidecar:
tags_.push_back({context_.reporter_, context_.source_});
break;
case Reporter::Unspecified:
// Unreachable: resolveReporter() never returns Unspecified.
IS_ENVOY_BUG("unresolved istio stats reporter");
tags_.push_back({context_.reporter_, context_.unknown_});
break;
}
}
~IstioStatsFilter() override { ASSERT(report_timer_ == nullptr); }

// Http::StreamDecoderFilter
void setDecoderFilterCallbacks(Http::StreamDecoderFilterCallbacks& callbacks) override {
decoder_callbacks_ = &callbacks;
initializeTags();
}
Http::FilterHeadersStatus decodeHeaders(Http::RequestHeaderMap& request_headers, bool) override {
is_grpc_ = Grpc::Common::isGrpcRequestHeaders(request_headers);
if (is_grpc_) {
Expand Down Expand Up @@ -866,6 +906,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,
}
void initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) override {
network_read_callbacks_ = &callbacks;
initializeTags();
network_read_callbacks_->connection().addConnectionCallbacks(*this);
}
// Network::ConnectionCallbacks
Expand Down Expand Up @@ -893,7 +934,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,
if (decoder_callbacks_) {
if (!peer_read_) {
const auto& info = decoder_callbacks_->streamInfo();
peer_read_ = peerInfoRead(config_->reporter(), info.filterState());
peer_read_ = peerInfoRead(reporter(), info.filterState());
if (peer_read_ || end_stream) {
ENVOY_LOG(trace, "Populating peer metadata from HTTP MX.");
populatePeerInfo(info, info.filterState());
Expand Down Expand Up @@ -922,7 +963,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,
const auto& info = network_read_callbacks_->connection().streamInfo();
// TCP MX writes to upstream stream info instead.
OptRef<const StreamInfo::UpstreamInfo> upstream_info;
if (config_->reporter() == Reporter::ClientSidecar) {
if (reporter() == Reporter::ClientSidecar) {
upstream_info = info.upstreamInfo();
}
const StreamInfo::FilterState& filter_state =
Expand All @@ -931,7 +972,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,
: info.filterState();

if (!peer_read_) {
peer_read_ = peerInfoRead(config_->reporter(), filter_state);
peer_read_ = peerInfoRead(reporter(), filter_state);
// Report connection open once peer info is read or connection is closed.
if (peer_read_ || end_stream) {
ENVOY_LOG(trace, "Populating peer metadata from TCP MX.");
Expand Down Expand Up @@ -979,10 +1020,10 @@ class IstioStatsFilter : public Http::PassThroughFilter,
const StreamInfo::FilterState& filter_state) {
// Compute peer info with client-side fallbacks.
std::optional<Istio::Common::WorkloadMetadataObject> peer;
auto object = peerInfo(config_->reporter(), filter_state);
auto object = peerInfo(reporter(), filter_state);
if (object) {
peer.emplace(object.value());
} else if (config_->reporter() == Reporter::ClientSidecar) {
} else if (reporter() == Reporter::ClientSidecar) {
if (auto label_obj = extractEndpointMetadata(info); label_obj) {
peer.emplace(label_obj.value());
}
Expand Down Expand Up @@ -1044,7 +1085,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,

std::string peer_san;
absl::string_view local_san;
switch (config_->reporter()) {
switch (reporter()) {
case Reporter::ServerSidecar:
case Reporter::ServerGateway: {
auto peer_principal =
Expand Down Expand Up @@ -1078,7 +1119,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,
peer_san = ssl_info->uriSanPeerCertificate()[0];
}
if (peer_san.empty()) {
auto endpoint_object = peerInfo(config_->reporter(), filter_state);
auto endpoint_object = peerInfo(reporter(), filter_state);
if (endpoint_object) {
endpoint_peer.emplace(endpoint_object.value());
peer_san = endpoint_peer->identity_;
Expand All @@ -1091,6 +1132,8 @@ class IstioStatsFilter : public Http::PassThroughFilter,
}
break;
}
case Reporter::Unspecified:
break;
}
// Implements fallback from using the namespace from SAN if available to
// using peer metadata, otherwise.
Expand All @@ -1104,7 +1147,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,
if (peer_namespace.empty() && peer) {
peer_namespace = peer->namespace_name_;
}
switch (config_->reporter()) {
switch (reporter()) {
case Reporter::ServerSidecar:
case Reporter::ServerGateway: {
tags_.push_back({context_.source_workload_, peer && !peer->workload_name_.empty()
Expand All @@ -1130,7 +1173,7 @@ class IstioStatsFilter : public Http::PassThroughFilter,
tags_.push_back({context_.source_cluster_, peer && !peer->cluster_name_.empty()
? pool_.add(peer->cluster_name_)
: context_.unknown_});
switch (config_->reporter()) {
switch (reporter()) {
case Reporter::ServerGateway: {
std::optional<Istio::Common::WorkloadMetadataObject> endpoint_peer;
auto endpoint_object = peerInfo(Reporter::ClientSidecar, filter_state);
Expand Down Expand Up @@ -1256,8 +1299,9 @@ class IstioStatsFilter : public Http::PassThroughFilter,
Context& context_;
Stats::StatNameDynamicPool pool_;
Stats::StatNameTagVector tags_;
Reporter reporter_{Reporter::ClientSidecar};
Event::TimerPtr report_timer_{nullptr};
Network::ReadFilterCallbacks* network_read_callbacks_;
Network::ReadFilterCallbacks* network_read_callbacks_{nullptr};
bool peer_read_{false};
uint64_t bytes_sent_{0};
uint64_t bytes_received_{0};
Expand All @@ -1272,12 +1316,12 @@ class IstioStatsFilter : public Http::PassThroughFilter,
} // namespace

absl::StatusOr<Http::FilterFactoryCb>
IstioStatsFilterConfigFactory::createFilterFactoryFromProtoTyped(
const stats::PluginConfig& proto_config, const std::string&,
Server::Configuration::FactoryContext& factory_context) {
factory_context.serverFactoryContext().api().customStatNamespaces().registerStatNamespace(
CustomStatNamespace);
ConfigSharedPtr config = std::make_shared<Config>(proto_config, factory_context);
IstioStatsFilterConfigFactory::createHttpFilterFactoryFromProtoTyped(
const stats::PluginConfig& proto_config, Server::Configuration::ServerFactoryContext& context,
Server::Configuration::ExtraFactoryContext& extra_context) {
context.api().customStatNamespaces().registerStatNamespace(CustomStatNamespace);
ConfigSharedPtr config =
std::make_shared<Config>(proto_config, context, extra_context.scopeOr(context));
return [config](Http::FilterChainFactoryCallbacks& callbacks) {
auto filter = std::make_shared<IstioStatsFilter>(config);
callbacks.addStreamFilter(filter);
Expand All @@ -1295,8 +1339,9 @@ IstioStatsNetworkFilterConfigFactory::createFilterFactoryFromProto(
const Protobuf::Message& proto_config, Server::Configuration::FactoryContext& factory_context) {
factory_context.serverFactoryContext().api().customStatNamespaces().registerStatNamespace(
CustomStatNamespace);
ConfigSharedPtr config = std::make_shared<Config>(
dynamic_cast<const stats::PluginConfig&>(proto_config), factory_context);
ConfigSharedPtr config =
std::make_shared<Config>(dynamic_cast<const stats::PluginConfig&>(proto_config),
factory_context.serverFactoryContext(), factory_context.scope());
return [config](Network::FilterManager& filter_manager) {
filter_manager.addReadFilter(std::make_shared<IstioStatsFilter>(config));
};
Expand Down
9 changes: 5 additions & 4 deletions contrib/istio/filters/http/istio_stats/source/istio_stats.h
Original file line number Diff line number Diff line change
Expand Up @@ -13,14 +13,15 @@ namespace Extensions {
namespace HttpFilters {
namespace IstioStats {

class IstioStatsFilterConfigFactory : public Common::ExceptionFreeFactoryBase<stats::PluginConfig> {
class IstioStatsFilterConfigFactory : public Common::UnifiedFactoryBase<stats::PluginConfig> {
public:
IstioStatsFilterConfigFactory() : ExceptionFreeFactoryBase("envoy.filters.http.istio_stats") {}
IstioStatsFilterConfigFactory() : UnifiedFactoryBase("envoy.filters.http.istio_stats") {}

private:
absl::StatusOr<Http::FilterFactoryCb>
createFilterFactoryFromProtoTyped(const stats::PluginConfig& proto_config, const std::string&,
Server::Configuration::FactoryContext&) override;
createHttpFilterFactoryFromProtoTyped(const stats::PluginConfig& proto_config,
Server::Configuration::ServerFactoryContext&,
Server::Configuration::ExtraFactoryContext&) override;
};

class IstioStatsNetworkFilterConfigFactory
Expand Down
Loading