apiVersion: networking.istio.io/v1alpha3kind: Gatewaymetadata:  name: my-gateway  namespace: defaultspec:  selector:    istio: ingressgateway  servers:  - port:      number: 80      name: http      protocol: HTTP    hosts:    - "bookinfo.example.com"---apiVersion: networking.istio.io/v1alpha3kind: VirtualServicemetadata:  name: bookinfo-vs  namespace: defaultspec:  hosts:  - "bookinfo.example.com"  gateways:  - my-gateway  http:  - match:    - uri:        prefix: /reviews    rewrite:      uri: /    timeout: 5s    retries:      attempts: 3      perTryTimeout: 2s      retryOn: "5xx,connect-failure"    route:    - destination:        host: reviews        port:          number: 9080        subset: v2      weight: 80    - destination:        host: reviews        port:          number: 9080        subset: v3      weight: 20

func(ps *PushContext) initGateways(env *Environment) error {    gatewayConfigs, err := env.List(gvk.Gateway, NamespaceAll)    sortConfigByCreationTime(gatewayConfigs)    ps.gatewayIndex.all = gatewayConfigs    ps.gatewayIndex.namespace = make(map[string][]config.Config)    for _, gatewayConfig := range gatewayConfigs {        ps.gatewayIndex.namespace[gatewayConfig.Namespace] = append(...)    }}

type MergedGateway struct {    ServerPorts          []ServerPort    MergedServers        map[ServerPort]*MergedServers    GatewayNameForServer map[*networking.Server]string      // Server → "default/my-gateway"    ServersByRouteName   map[string][]*networking.Server    // "http.80" → [Server]    TLSServerInfo        map[*networking.Server]*TLSServerInfo}

func(configgen *ConfigGeneratorImpl) buildGatewayListeners(builder *ListenerBuilder) {    for _, port := range mergedGateway.ServerPorts {        // 协议判断 → 构建 FilterChain        if p.IsHTTP() {            // 单个 FilterChain，所有同端口 HTTP Server 共享            opts.filterChainOpts = []*filterChainOpts{                configgen.createGatewayHTTPFilterChainOpts(node, port, nil,                    serversForPort.RouteName, ...)  // RouteName = "http.80"            }        }    }}

funcbuildGatewayHTTPRouteConfig(node, push, routeName) *route.RouteConfiguration {    servers := merged.ServersByRouteName[routeName]  // "http.80" → Servers    for _, server := range servers {        gatewayName := merged.GatewayNameForServer[server]       // "default/my-gateway"        virtualServices := push.VirtualServicesForGateway(node, gatewayName) // 找绑定的 VS        for _, vs := range virtualServices {            // 计算 Host 交集            intersectingHosts := serverHosts.Intersection(virtualServiceHosts)            // 构建 Envoy Route            routes := istio_route.BuildHTTPRoutesForVirtualService(...)            // 创建 VirtualHost            newVHost := &route.VirtualHost{                Name:    "bookinfo.example.com:80",                Domains: ["bookinfo.example.com", "bookinfo.example.com:80"],                Routes:  routes,            }        }    }}

// uri.prefix: /reviews → RouteMatch.Prefix = "/reviews"case *networking.StringMatch_Prefix:    out.PathSpecifier = &route.RouteMatch_Prefix{Prefix: m.Prefix}

if in.Rewrite != nil {    action.PrefixRewrite = in.Rewrite.GetUri()  // "/"

if in.Timeout != nil {    d = gogo.DurationToProtoDuration(in.Timeout)  // 5s}action.Timeout = d

func ConvertPolicy(in *networking.HTTPRetry) *route.RetryPolicy {    out := DefaultPolicy()  // 基础：2次, connect-failure,refused-stream,unavailable,cancelled    out.NumRetries = uint32(in.Attempts)   // 3    out.RetryOn = parseRetryOn(in.RetryOn) // "5xx,connect-failure"    out.PerTryTimeout = in.PerTryTimeout   // 2s    // RetryHostPredicate: previous_hosts（避免重试同一台机器）}

for _, dst := range in.Route {    n := GetDestinationCluster(dst.Destination, ...)    // → "outbound|9080|v2|reviews.default.svc.cluster.local"    weighted = append(weighted, &route.WeightedCluster_ClusterWeight{Name: n, Weight: dst.Weight})}action.ClusterSpecifier = &route.RouteAction_WeightedClusters{WeightedClusters: ...}

```goif features.FilterGatewayClusterConfig && proxy.Type == model.Router {    services = req.Push.GatewayServices(proxy)  // 只取 VS 引用的}

var PushOrder = []string{ClusterType, EndpointType, ListenerType, RouteType, SecretType}funcpushConnection(con, pushEv) error {    for _, w := range orderWatchedResources(con.proxy.WatchedResources) {        s.pushXds(con, pushRequest.Push, w, pushRequest)    }}

var PushOrder = []string{ClusterType, EndpointType, ListenerType, RouteType, SecretType}funcpushConnection(con, pushEv) error {    for _, w := range orderWatchedResources(con.proxy.WatchedResources) {        s.pushXds(con, pushRequest.Push, w, pushRequest)    }}

// source/common/config/grpc_mux_impl.cc:104-130GrpcMuxWatchPtr GrpcMuxImpl::addWatch(conststd::string& type_url, ...) {    if (!apiStateFor(type_url).subscribed_) {        apiStateFor(type_url).subscribed_ = true;        subscriptions_.emplace_back(type_url);  // 按首次订阅顺序入队    }    queueDiscoveryRequest(type_url);}

// source/common/config/grpc_mux_impl.cc:291-296void GrpcMuxImpl::onStreamEstablished() {    for (const auto& type_url : subscriptions_) {        queueDiscoveryRequest(type_url);  // CDS → EDS → LDS → RDS 顺序    }}

// source/common/config/grpc_mux_impl.cc:158-284void GrpcMuxImpl::onDiscoveryResponse(unique_ptr<DiscoveryResponse>&& message, ...) {    // 解码资源    for(const auto& resource : message->resources()) {        decoded_resource = DecodedResourceImpl::fromResource(...);        resources.emplace_back(std::move(decoded_resource));    }    // 分发给订阅回调    for(auto watch : api_state.watches_) {        watch->callbacks_.onConfigUpdate(found_resources, message->version_info());    }    // ACK：更新 version_info    api_state.request_.set_version_info(message->version_info());    api_state.request_.set_response_nonce(message->nonce());    queueDiscoveryRequest(type_url);  // 发送 ACK}// NACK：catch 到异常时不更新 version_info，设置 error_detailcatch(const EnvoyException& e) {    error_detail->set_code(Grpc::Status::WellKnownGrpcStatus::Internal);    error_detail->set_message(e.what());}

// source/common/upstream/cluster_manager_impl.cc:839-863// 有集群在 warming 时，暂停 CDS ACKif(previous_warming == 0 && !warming_clusters_.empty()) {    resume_cds_ = ads_mux_->pause(type_url);  // 暂停 CDS}// 所有集群 warm 后，恢复 CDS，ACK 发往 ADS// ADS 据此知道集群已就绪，可以推 RDSelse if (previous_warming > 0 && warming_clusters_.empty()) {    resume_cds_.reset();  // 恢复}

// source/common/upstream/cds_api_impl.cc:41-101voidCdsApiImpl::onConfigUpdate(added_resources, removed_resources, version_info){    // 更新时暂停 EDS/LEDS    maybe_resume_eds = cm_.adsMux()->pause(eds_type_url);    for (const auto& resource : added_resources) {        cm_.addOrUpdateCluster(cluster, version);  // 添加/更新集群    }    for (const auto& removed : removed_resources) {        cm_.removeCluster(removed);    }}

// source/common/upstream/eds.cc:46-127void EdsClusterImpl::BatchUpdateHelper::batchUpdate(host_update_cb) {    PriorityStateManager priority_state_manager(...);    for (const auto& locality_lb_endpoint : cluster_load_assignment_.endpoints()) {        priority_state_manager.initializePriorityFor(locality_lb_endpoint);        for (const auto& lb_endpoint : locality_lb_endpoint.lb_endpoints()) {            updateLocalityEndpoints(lb_endpoint, locality_lb_endpoint,                                    priority_state_manager, all_new_hosts);        }    }    // 更新 PrioritySet，触发 LB 重建}

// source/server/lds_api.cc:41-101voidLdsApiImpl::onConfigUpdate(added_resources, removed_resources, version_info){    // 更新 Listener 时暂停 RDS    maybe_resume_rds = cm_.adsMux()->pause(route_type_url);    // 先删除    for (const auto& removed : removed_resources) {        listener_manager_.removeListener(removed);    }    // 再添加/更新    for (const auto& resource : added_resources) {        listener_manager_.addOrUpdateListener(listener, version, true);    }}

// source/common/router/rds_impl.cc:112-162void RdsRouteConfigSubscription::onConfigUpdate(resources, version_info) {    config_update_info_->onRdsUpdate(route_config, version_info);    route_config_provider_->onConfigUpdate();}// source/common/router/rds_impl.cc:262-294void RdsRouteConfigProviderImpl::onConfigUpdate() {    // 通过 ThreadLocal 原子替换路由配置到所有 worker 线程    tls_.runOnAllThreads([new_config = config_update_info_->parsedConfiguration()](                             OptRef<ThreadLocalConfig> tls) {        tls->config_ = new_config;    });}

客户端请求 → Listener (0.0.0.0:8080)  → TcpListenerImpl::onAccept()  → ActiveTcpListener::onAcceptWorker()    → connectionBalancer().pickTargetHandler()  // 分配到某个 worker  → ActiveTcpSocket::continueFilterChain()    → 执行 listener filter (如 TLS Inspector)  → newConnection()    → FilterChainManager::findFilterChain(socket)  // 匹配 FilterChain    → 创建网络连接，安装 network filter（HCM 等）

HCM::onData(data)  → codec_->dispatch(data)   // HTTP/1.x 或 H2 解码  → newStream() → ActiveStream  → decodeHeaders(headers)    → snapped_route_config_ = routeConfigProvider()->config()  // 快照当前路由配置    → refreshCachedRoute()      → snapped_route_config_->route(headers, stream_info)        → RouteMatcher::route()          → findVirtualHost(headers)  // Host 头匹配          → VirtualHostImpl::getRouteFromEntries()  // 按顺序匹配 Route

findVirtualHost(headers):  1. headers.getHostValue() → "bookinfo.example.com"  2. virtual_hosts_.find("bookinfo.example.com")    // 精确匹配 ✓  3. wildcard_virtual_host_suffixes_                 // 后缀通配 (*.example.com)  4. wildcard_virtual_host_prefixes_                 // 前缀通配  5. default_virtual_host_                           // 兜底

getRouteFromEntries(headers):  遍历 routes_，对每个 route 调用 matches():    → PrefixRouteEntryImpl::matches()      → path.startsWith("/reviews")  ✓      → 检查 headers matcher      → 检查 query params      → 返回匹配的 RouteEntry

// config_impl.cc:735-746absl::optional<std::string> currentUrlPathAfterRewriteWithMatchedPath(headers, matched_path){    const auto& rewrite = getPathRewrite(headers);  // "/"    std::string path(headers.getPathValue());         // "/reviews/detail?id=1"    return path.replace(0, matched_path.size(), rewrite);    // matched_path = "/reviews", rewrite = "/"    // 结果: "/detail?id=1"}

// pickWeightedCluster():selected_value = random_value % total_cluster_weight_;  // total = 100// 遍历 weighted_clusters_://   v2: [0, 80)  → 80% 概率//   v3: [80, 100) → 20% 概率for (const auto& cluster : weighted_clusters_) {    selected_value -= cluster.clusterWeight();    if (selected_value <= 0) {        return cluster;  // 选中    }}

Router::Filter::decodeHeaders()  → createConnPool()    → cm_.getThreadLocalCluster("outbound|9080|v2|reviews.default.svc.cluster.local")    → lb_->chooseHost(context)      → chooseHostSet()       // 按 priority 选 HostSet      → chooseHostOnce()      // 在 HostSet 中做 LB（RR/Random/LeastRequest 等）      → 返回 Host: 10.244.1.5:9080

Router::Filter::decodeHeaders()  → 创建 UpstreamRequest  → upstream_request->encodeHeaders(end_stream)    → conn_pool_->newStream(this)   // 从连接池获取/创建连接    → onPoolReady(encoder, host)    // 连接就绪    → encoder->encodeHeaders(...)   // 发送请求头到上游    → encoder->encodeData(...)      // 发送请求体

// router.cc:661retry_state_ = createRetryState(route_entry_->retryPolicy(), headers, *cluster_, ...);// RetryPolicy 来自 RDS 配置://   num_retries=3, retry_on="5xx,connect-failure", per_try_timeout=2s

上游返回 503:  Filter::onUpstreamHeaders()    → retry_state_->shouldRetryHeaders(response_headers, ...)      → wouldRetryFromHeaders()  // 检查 5xx → true      → 检查 budget/circuit breaker      → 返回 RetryStatus::Yes连接失败:  Filter::onUpstreamReset()    → retry_state_->shouldRetryReset(reset_reason, ...)      → wouldRetryFromReset()  // 检查 connect-failure → truePer-try 超时 (2s):  UpstreamRequest::onPerTryTimeout()    → parent_.onPerTryTimeout()      → retry_state_->shouldRetryReset(timeout, ...)

// router.cc:1666-1730voidFilter::doRetry(bool can_send_early_data, bool can_use_http3, TimeoutRetry is_timeout_retry){    // 创建新的 UpstreamRequest    auto upstream_request = std::make_unique<UpstreamRequest>(*this, ...);    upstream_request->is_retry_ = true;    // RetryHostPredicate: previous_hosts → 避免选择已失败的 host    upstream_request->encodeHeaders(end_stream);}

请求到达 → 选 host A → 发送 → 503 (耗时 200ms)  → 重试 1: 选 host B (避开 A) → 发送 → connect-failure (耗时 100ms)  → 重试 2: 选 host C (避开 A,B) → 发送 → 200 OK总耗时 < 5s (全局 timeout)每次重试 < 2s (per_try_timeout)

// FilterUtility::finalTimeout():global_timeout = route.timeout()               // 5s (来自 VS 的 timeout)per_try_timeout = route.retryPolicy().perTryTimeout()  // 2s

全局超时 (5s):  Filter::onRequestComplete()    → response_timeout_ = dispatcher.createTimer(onResponseTimeout)    → response_timeout_->enableTimer(5s)  超时触发:  Filter::onResponseTimeout()    → 重置所有 upstream 请求    → 返回 504 Gateway TimeoutPer-try 超时 (2s):  UpstreamRequest::setupPerTryTimeout()    → per_try_timeout_->enableTimer(2s)  超时触发:  UpstreamRequest::onPerTryTimeout()    → 触发 retry（如果还有重试次数）    → 或返回 504