Grcov report - conn_manager

1

#include "source/common/http/conn_manager_impl.h"

2

3

#include <chrono>

4

#include <cstdint>

5

#include <functional>

6

#include <iterator>

7

#include <limits>

8

#include <list>

9

#include <memory>

10

#include <string>

11

#include <vector>

12

13

#include "envoy/buffer/buffer.h"

14

#include "envoy/common/time.h"

15

#include "envoy/config/core/v3/base.pb.h"

16

#include "envoy/event/dispatcher.h"

17

#include "envoy/event/scaled_range_timer_manager.h"

18

#include "envoy/extensions/filters/network/http_connection_manager/v3/http_connection_manager.pb.h"

19

#include "envoy/http/header_map.h"

20

#include "envoy/http/header_validator_errors.h"

21

#include "envoy/network/drain_decision.h"

22

#include "envoy/router/router.h"

23

#include "envoy/ssl/connection.h"

24

#include "envoy/stats/scope.h"

25

#include "envoy/stream_info/filter_state.h"

26

#include "envoy/stream_info/stream_info.h"

27

#include "envoy/tracing/tracer.h"

28

#include "envoy/type/v3/percent.pb.h"

29

30

#include "source/common/buffer/buffer_impl.h"

31

#include "source/common/common/assert.h"

32

#include "source/common/common/empty_string.h"

33

#include "source/common/common/enum_to_int.h"

34

#include "source/common/common/fmt.h"

35

#include "source/common/common/perf_tracing.h"

36

#include "source/common/common/scope_tracker.h"

37

#include "source/common/common/utility.h"

38

#include "source/common/http/codes.h"

39

#include "source/common/http/conn_manager_utility.h"

40

#include "source/common/http/exception.h"

41

#include "source/common/http/header_map_impl.h"

42

#include "source/common/http/header_utility.h"

43

#include "source/common/http/headers.h"

44

#include "source/common/http/http1/codec_impl.h"

45

#include "source/common/http/http2/codec_impl.h"

46

#include "source/common/http/path_utility.h"

47

#include "source/common/http/status.h"

48

#include "source/common/http/utility.h"

49

#include "source/common/network/utility.h"

50

#include "source/common/router/config_impl.h"

51

#include "source/common/runtime/runtime_features.h"

52

#include "source/common/stats/timespan_impl.h"

53

#include "source/common/stream_info/utility.h"

54

55

#include "absl/strings/escaping.h"

56

#include "absl/strings/match.h"

57

#include "absl/strings/str_cat.h"

58

59

namespace Envoy {

60

namespace Http {

61

62

const absl::string_view ConnectionManagerImpl::PrematureResetTotalStreamCountKey =

63

    "overload.premature_reset_total_stream_count";

64

const absl::string_view ConnectionManagerImpl::PrematureResetMinStreamLifetimeSecondsKey =

65

    "overload.premature_reset_min_stream_lifetime_seconds";

66

// Runtime key for maximum number of requests that can be processed from a single connection per

67

// I/O cycle. Requests over this limit are deferred until the next I/O cycle.

68

const absl::string_view ConnectionManagerImpl::MaxRequestsPerIoCycle =

69

    "http.max_requests_per_io_cycle";

70

// Don't attempt to intelligently delay close: https://github.com/envoyproxy/envoy/issues/30010

71

const absl::string_view ConnectionManagerImpl::OptionallyDelayClose =

72

    "http1.optionally_delay_close";

73

74

2805

bool requestWasConnect(const RequestHeaderMapSharedPtr& headers, Protocol protocol) {

75

2805

  if (!headers) {

76

1349

    return false;

77

1349

78

1456

  if (protocol <= Protocol::Http11) {

79

547

    return HeaderUtility::isConnect(*headers);

80

547

81

  // All HTTP/2 style upgrades were originally connect requests.

82

909

  return HeaderUtility::isConnect(*headers) || Utility::isUpgrade(*headers);

83

1456

84

85

const Formatter::Formatter*

86

operationNameFormatter(const Http::TracingConnectionManagerConfig& hcm_config,

87

158

                       const Router::RouteTracing* route_config) {

88

158

  const Formatter::Formatter* formatter =

89

158

      route_config != nullptr ? route_config->operation().ptr() : nullptr;

90

158

  return formatter != nullptr ? formatter : hcm_config.operation_.get();

91

158

92

const Formatter::Formatter*

93

upstreamOperationNameFormatter(const Http::TracingConnectionManagerConfig& hcm_config,

94

17

                               const Router::RouteTracing* route_config) {

95

17

  const Formatter::Formatter* formatter =

96

17

      route_config != nullptr ? route_config->upstreamOperation().ptr() : nullptr;

97

17

  return formatter != nullptr ? formatter : hcm_config.upstream_operation_.get();

98

17

99

100

ConnectionManagerStats ConnectionManagerImpl::generateStats(const std::string& prefix,

101

20710

                                                            Stats::Scope& scope) {

102

20710

  return ConnectionManagerStats(

103

20710

      {ALL_HTTP_CONN_MAN_STATS(POOL_COUNTER_PREFIX(scope, prefix), POOL_GAUGE_PREFIX(scope, prefix),

104

20710

                               POOL_HISTOGRAM_PREFIX(scope, prefix))},

105

20710

      prefix, scope);

106

20710

107

108

ConnectionManagerTracingStats ConnectionManagerImpl::generateTracingStats(const std::string& prefix,

109

20710

                                                                          Stats::Scope& scope) {

110

20710

  return {CONN_MAN_TRACING_STATS(POOL_COUNTER_PREFIX(scope, prefix + "tracing."))};

111

20710

112

113

ConnectionManagerListenerStats

114

20672

ConnectionManagerImpl::generateListenerStats(const std::string& prefix, Stats::Scope& scope) {

115

20672

  return {CONN_MAN_LISTENER_STATS(POOL_COUNTER_PREFIX(scope, prefix))};

116

20672

117

118

ConnectionManagerImpl::ConnectionManagerImpl(

119

    ConnectionManagerConfigSharedPtr config, const Network::DrainDecision& drain_close,

120

    Random::RandomGenerator& random_generator, Http::Context& http_context,

121

    Runtime::Loader& runtime, const LocalInfo::LocalInfo& local_info,

122

    Upstream::ClusterManager& cluster_manager, Server::OverloadManager& overload_manager,

123

    TimeSource& time_source, envoy::config::core::v3::TrafficDirection direction)

124

22431

    : config_(std::move(config)), stats_(config_->stats()),

125

22431

      conn_length_(new Stats::HistogramCompletableTimespanImpl(

126

22431

          stats_.named_.downstream_cx_length_ms_, time_source)),

127

22431

      drain_close_(drain_close), user_agent_(http_context.userAgentContext()),

128

22431

      random_generator_(random_generator), runtime_(runtime), local_info_(local_info),

129

22431

      cluster_manager_(cluster_manager), listener_stats_(config_->listenerStats()),

130

22431

      overload_manager_(overload_manager),

131

22431

      overload_state_(overload_manager.getThreadLocalOverloadState()),

132

      accept_new_http_stream_(

133

22431

          overload_manager.getLoadShedPoint(Server::LoadShedPointName::get().HcmDecodeHeaders)),

134

      hcm_ondata_creating_codec_(

135

22431

          overload_manager.getLoadShedPoint(Server::LoadShedPointName::get().HcmCodecCreation)),

136

      overload_stop_accepting_requests_ref_(

137

22431

          overload_state_.getState(Server::OverloadActionNames::get().StopAcceptingRequests)),

138

      overload_disable_keepalive_ref_(

139

22431

          overload_state_.getState(Server::OverloadActionNames::get().DisableHttpKeepAlive)),

140

22431

      time_source_(time_source), proxy_name_(StreamInfo::ProxyStatusUtils::makeProxyName(

141

22431

                                     /*node_id=*/local_info_.node().id(),

142

22431

                                     /*server_name=*/config_->serverName(),

143

22431

                                     /*proxy_status_config=*/config_->proxyStatusConfig())),

144

      max_requests_during_dispatch_(

145

22431

          runtime_.snapshot().getInteger(ConnectionManagerImpl::MaxRequestsPerIoCycle, UINT32_MAX)),

146

22431

      direction_(direction),

147

22431

      allow_upstream_half_close_(Runtime::runtimeFeatureEnabled(

148

22431

          "envoy.reloadable_features.allow_multiplexed_upstream_half_close")),

149

22431

      close_connection_on_zombie_stream_complete_(Runtime::runtimeFeatureEnabled(

150

22431

          "envoy.reloadable_features.http1_close_connection_on_zombie_stream_complete")) {

151

22431

  ENVOY_LOG_ONCE_IF(

152

22431

      trace, accept_new_http_stream_ == nullptr,

153

22431

      "LoadShedPoint envoy.load_shed_points.http_connection_manager_decode_headers is not "

154

22431

      "found. Is it configured?");

155

22431

  ENVOY_LOG_ONCE_IF(trace, hcm_ondata_creating_codec_ == nullptr,

156

22431

                    "LoadShedPoint envoy.load_shed_points.hcm_ondata_creating_codec is not found. "

157

22431

                    "Is it configured?");

158

22431

159

160

80

const ResponseHeaderMap& ConnectionManagerImpl::continueHeader() {

161

80

  static const auto headers = createHeaderMap<ResponseHeaderMapImpl>(

162

80

      {{Http::Headers::get().Status, std::to_string(enumToInt(Code::Continue))}});

163

80

  return *headers;

164

80

165

166

22431

void ConnectionManagerImpl::initializeReadFilterCallbacks(Network::ReadFilterCallbacks& callbacks) {

167

22431

  read_callbacks_ = &callbacks;

168

22431

  dispatcher_ = &callbacks.connection().dispatcher();

169

22431

  if (max_requests_during_dispatch_ != UINT32_MAX) {

170

22

    deferred_request_processing_callback_ =

171

324

        dispatcher_->createSchedulableCallback([this]() -> void { onDeferredRequestProcessing(); });

172

22

173

174

22431

  stats_.named_.downstream_cx_total_.inc();

175

22431

  stats_.named_.downstream_cx_active_.inc();

176

22431

  if (read_callbacks_->connection().ssl()) {

177

2202

    stats_.named_.downstream_cx_ssl_total_.inc();

178

2202

    stats_.named_.downstream_cx_ssl_active_.inc();

179

2202

180

181

22431

  read_callbacks_->connection().addConnectionCallbacks(*this);

182

183

22431

  if (config_->addProxyProtocolConnectionState() &&

184

22431

      !read_callbacks_->connection()

185

22430

           .streamInfo()

186

22430

           .filterState()

187

22430

           ->hasData<Network::ProxyProtocolFilterState>(Network::ProxyProtocolFilterState::key())) {

188

22421

    read_callbacks_->connection().streamInfo().filterState()->setData(

189

22421

        Network::ProxyProtocolFilterState::key(),

190

22421

        std::make_unique<Network::ProxyProtocolFilterState>(Network::ProxyProtocolData{

191

22421

            read_callbacks_->connection().connectionInfoProvider().remoteAddress(),

192

22421

            read_callbacks_->connection().connectionInfoProvider().localAddress()}),

193

22421

        StreamInfo::FilterState::StateType::ReadOnly,

194

22421

        StreamInfo::FilterState::LifeSpan::Connection);

195

22421

196

197

22431

  if (config_->idleTimeout()) {

198

21757

    connection_idle_timer_ =

199

21757

        dispatcher_->createScaledTimer(Event::ScaledTimerType::HttpDownstreamIdleConnectionTimeout,

200

21757

                                       [this]() -> void { onIdleTimeout(); });

201

21757

    connection_idle_timer_->enableTimer(config_->idleTimeout().value());

202

21757

203

204

22431

  if (config_->maxConnectionDuration()) {

205

85

    connection_duration_timer_ =

206

85

        dispatcher_->createScaledTimer(Event::ScaledTimerType::HttpDownstreamMaxConnectionTimeout,

207

85

                                       [this]() -> void { onConnectionDurationTimeout(); });

208

85

    connection_duration_timer_->enableTimer(config_->maxConnectionDuration().value());

209

85

210

211

22431

  read_callbacks_->connection().setDelayedCloseTimeout(config_->delayedCloseTimeout());

212

213

22431

  read_callbacks_->connection().setConnectionStats(

214

22431

      {stats_.named_.downstream_cx_rx_bytes_total_, stats_.named_.downstream_cx_rx_bytes_buffered_,

215

22431

       stats_.named_.downstream_cx_tx_bytes_total_, stats_.named_.downstream_cx_tx_bytes_buffered_,

216

22431

       nullptr, &stats_.named_.downstream_cx_delayed_close_timeout_});

217

22431

218

219

22431

ConnectionManagerImpl::~ConnectionManagerImpl() {

220

22431

  stats_.named_.downstream_cx_destroy_.inc();

221

222

22431

  stats_.named_.downstream_cx_active_.dec();

223

22431

  if (read_callbacks_->connection().ssl()) {

224

2202

    stats_.named_.downstream_cx_ssl_active_.dec();

225

2202

226

227

22431

  if (codec_) {

228

22290

    if (codec_->protocol() == Protocol::Http2) {

229

7993

      stats_.named_.downstream_cx_http2_active_.dec();

230

18046

    } else if (codec_->protocol() == Protocol::Http3) {

231

1995

      stats_.named_.downstream_cx_http3_active_.dec();

232

13022

    } else {

233

12302

      stats_.named_.downstream_cx_http1_active_.dec();

234

12302

235

22290

236

237

22431

  if (soft_drain_http1_) {

238

9

    stats_.named_.downstream_cx_http1_soft_drain_.dec();

239

9

240

241

22431

  conn_length_->complete();

242

22431

  user_agent_.completeConnectionLength(*conn_length_);

243

22431

244

245

117866

void ConnectionManagerImpl::checkForDeferredClose(bool skip_delay_close) {

246

117866

  Network::ConnectionCloseType close = Network::ConnectionCloseType::FlushWriteAndDelay;

247

117866

  if (runtime_.snapshot().getBoolean(ConnectionManagerImpl::OptionallyDelayClose, true) &&

248

117866

      skip_delay_close) {

249

105

    close = Network::ConnectionCloseType::FlushWrite;

250

105

251

117866

  if (drain_state_ == DrainState::Closing && streams_.empty() && !codec_->wantsToWrite()) {

252

    // We are closing a draining connection with no active streams and the codec has

253

    // nothing to write.

254

2457

    doConnectionClose(close, absl::nullopt,

255

2457

                      StreamInfo::LocalCloseReasons::get().DeferredCloseOnDrainedConnection);

256

2457

257

117866

258

259

47242

void ConnectionManagerImpl::doEndStream(ActiveStream& stream, bool check_for_deferred_close) {

260

  // The order of what happens in this routine is important and a little complicated. We first see

261

  // if the stream needs to be reset. If it needs to be, this will end up invoking reset callbacks

262

  // and then moving the stream to the deferred destruction list. If the stream has not been reset,

263

  // we move it to the deferred deletion list here. Then, we potentially close the connection. This

264

  // must be done after deleting the stream since the stream refers to the connection and must be

265

  // deleted first.

266

47242

  bool reset_stream = false;

267

  // If the response encoder is still associated with the stream, reset the stream. The exception

268

  // here is when Envoy "ends" the stream by calling recreateStream at which point recreateStream

269

  // explicitly nulls out response_encoder to avoid the downstream being notified of the

270

  // Envoy-internal stream instance being ended.

271

47242

  if (stream.response_encoder_ != nullptr &&

272

47242

      (!stream.filter_manager_.hasLastDownstreamByteReceived() ||

273

46946

       !stream.state_.codec_saw_local_complete_)) {

274

    // Indicate local is complete at this point so that if we reset during a continuation, we don't

275

    // raise further data or trailers.

276

2792

    ENVOY_STREAM_LOG(debug, "doEndStream() resetting stream", stream);

277

    // TODO(snowp): This call might not be necessary, try to clean up + remove setter function.

278

2792

    stream.filter_manager_.setLocalComplete();

279

2792

    stream.state_.codec_saw_local_complete_ = true;

280

281

    // Per https://tools.ietf.org/html/rfc7540#section-8.3 if there was an error

282

    // with the TCP connection during a CONNECT request, it should be

283

    // communicated via CONNECT_ERROR

284

2792

    if (requestWasConnect(stream.request_headers_, codec_->protocol()) &&

285

2792

        (stream.filter_manager_.streamInfo().hasResponseFlag(

286

166

             StreamInfo::CoreResponseFlag::UpstreamConnectionFailure) ||

287

166

         stream.filter_manager_.streamInfo().hasResponseFlag(

288

166

             StreamInfo::CoreResponseFlag::UpstreamConnectionTermination))) {

289

63

      stream.response_encoder_->getStream().resetStream(StreamResetReason::ConnectError);

290

2773

    } else {

291

2729

      const bool reset_with_error =

292

2729

          Runtime::runtimeFeatureEnabled("envoy.reloadable_features.reset_with_error");

293

2729

      if (stream.filter_manager_.streamInfo().hasResponseFlag(

294

2729

              StreamInfo::CoreResponseFlag::UpstreamProtocolError) &&

295

2729

          !Runtime::runtimeFeatureEnabled(

296

39

              "envoy.reloadable_features.reset_ignore_upstream_reason")) {

297

2

        stream.response_encoder_->getStream().resetStream(StreamResetReason::ProtocolError);

298

2727

      } else if (reset_with_error && stream.filter_manager_.streamInfo().hasResponseFlag(

299

2723

                                         StreamInfo::CoreResponseFlag::DownstreamProtocolError)) {

300

15

        stream.response_encoder_->getStream().resetStream(StreamResetReason::ProtocolError);

301

2712

      } else {

302

2712

        stream.response_encoder_->getStream().resetStream(StreamResetReason::LocalReset);

303

2712

304

2729

305

2792

    reset_stream = true;

306

2792

307

308

47242

  if (!reset_stream) {

309

44450

    doDeferredStreamDestroy(stream);

310

44450

311

312

47242

  if (reset_stream && codec_->protocol() < Protocol::Http2) {

313

1890

    drain_state_ = DrainState::Closing;

314

1890

315

316

47242

  if (check_for_deferred_close) {

317

46946

    checkForDeferredClose(stream.shouldSkipDeferredCloseDelay());

318

46946

319

47242

320

321

97061

void ConnectionManagerImpl::doDeferredStreamDestroy(ActiveStream& stream) {

322

97061

  if (!stream.state_.is_internally_destroyed_) {

323

96765

    ++closed_non_internally_destroyed_requests_;

324

96765

    if (isPrematureRstStream(stream)) {

325

23052

      ++number_premature_stream_resets_;

326

23052

327

96765

328

97061

  if (stream.max_stream_duration_timer_ != nullptr) {

329

45

    stream.max_stream_duration_timer_->disableTimer();

330

45

    stream.max_stream_duration_timer_ = nullptr;

331

45

332

97061

  if (stream.stream_idle_timer_ != nullptr) {

333

93297

    stream.stream_idle_timer_->disableTimer();

334

93297

    stream.stream_idle_timer_ = nullptr;

335

93297

336

97061

  stream.filter_manager_.disarmRequestTimeout();

337

97061

  if (stream.request_header_timer_ != nullptr) {

338

1

    stream.request_header_timer_->disableTimer();

339

1

    stream.request_header_timer_ = nullptr;

340

1

341

97061

  if (stream.access_log_flush_timer_ != nullptr) {

342

10

    stream.access_log_flush_timer_->disableTimer();

343

10

    stream.access_log_flush_timer_ = nullptr;

344

10

345

346

  // Only destroy the active stream if the underlying codec has notified us of

347

  // completion or we've internal redirect the stream.

348

97061

  if (!stream.canDestroyStream()) {

349

    // Track that this stream is not expecting any additional calls apart from

350

    // codec notification.

351

2996

    stream.state_.is_zombie_stream_ = true;

352

2996

    return;

353

2996

354

355

94065

  if (stream.response_encoder_ != nullptr) {

356

93769

    stream.response_encoder_->getStream().registerCodecEventCallbacks(nullptr);

357

93769

358

359

94065

  stream.completeRequest();

360

94065

  stream.filter_manager_.onStreamComplete();

361

362

  // For HTTP/3, skip access logging here and add deferred logging info

363

  // to stream info for QuicStatsGatherer to use later.

364

94065

  if (codec_ && codec_->protocol() == Protocol::Http3 &&

365

      // There was a downstream reset, log immediately.

366

94065

      !stream.filter_manager_.sawDownstreamReset() &&

367

      // On recreate stream, log immediately.

368

94065

      stream.response_encoder_ != nullptr &&

369

94065

      Runtime::runtimeFeatureEnabled(

370

1509

          "envoy.reloadable_features.quic_defer_logging_to_ack_listener")) {

371

1508

    stream.deferHeadersAndTrailers();

372

92703

  } else {

373

    // For HTTP/1 and HTTP/2, log here as usual.

374

92557

    stream.log(AccessLog::AccessLogType::DownstreamEnd);

375

92557

376

377

94065

  stream.filter_manager_.destroyFilters();

378

379

94065

  dispatcher_->deferredDelete(stream.removeFromList(streams_));

380

381

  // The response_encoder should never be dangling (unless we're destroying a

382

  // stream we are recreating) as the codec level stream will either outlive the

383

  // ActiveStream, or be alive in deferred deletion queue at this point.

384

94065

  if (stream.response_encoder_) {

385

93769

    stream.response_encoder_->getStream().removeCallbacks(stream);

386

93769

387

388

94065

  if (connection_idle_timer_ && streams_.empty()) {

389

27574

    connection_idle_timer_->enableTimer(config_->idleTimeout().value());

390

27574

391

94065

  maybeDrainDueToPrematureResets();

392

94065

393

394

RequestDecoderHandlePtr ConnectionManagerImpl::newStreamHandle(ResponseEncoder& response_encoder,

395

6

                                                               bool is_internally_created) {

396

6

  RequestDecoder& decoder = newStream(response_encoder, is_internally_created);

397

6

  return std::make_unique<ActiveStreamHandle>(static_cast<ActiveStream&>(decoder));

398

6

399

400

RequestDecoder& ConnectionManagerImpl::newStream(ResponseEncoder& response_encoder,

401

94065

                                                 bool is_internally_created) {

402

94065

  TRACE_EVENT("core", "ConnectionManagerImpl::newStream");

403

94065

  if (connection_idle_timer_) {

404

93282

    connection_idle_timer_->disableTimer();

405

93282

406

407

94065

  ENVOY_CONN_LOG(debug, "new stream", read_callbacks_->connection());

408

409

94065

  Buffer::BufferMemoryAccountSharedPtr downstream_stream_account =

410

94065

      response_encoder.getStream().account();

411

412

94065

  if (downstream_stream_account == nullptr) {

413

    // Create account, wiring the stream to use it for tracking bytes.

414

    // If tracking is disabled, the wiring becomes a NOP.

415

94061

    auto& buffer_factory = dispatcher_->getWatermarkFactory();

416

94061

    downstream_stream_account = buffer_factory.createAccount(response_encoder.getStream());

417

94061

    response_encoder.getStream().setAccount(downstream_stream_account);

418

94061

419

420

94065

  auto new_stream = std::make_unique<ActiveStream>(

421

94065

      *this, response_encoder.getStream().bufferLimit(), std::move(downstream_stream_account));

422

423

94065

  accumulated_requests_++;

424

94065

  if (config_->maxRequestsPerConnection() > 0 &&

425

94065

      accumulated_requests_ >= config_->maxRequestsPerConnection()) {

426

54

    if (codec_->protocol() < Protocol::Http2) {

427

22

      new_stream->filter_manager_.streamInfo().setShouldDrainConnectionUponCompletion(true);

428

      // Prevent erroneous debug log of closing due to incoming connection close header.

429

22

      drain_state_ = DrainState::Closing;

430

40

    } else if (drain_state_ == DrainState::NotDraining) {

431

21

      startDrainSequence();

432

21

433

54

    ENVOY_CONN_LOG(debug, "max requests per connection reached", read_callbacks_->connection());

434

54

    stats_.named_.downstream_cx_max_requests_reached_.inc();

435

54

436

437

94065

  if (soft_drain_http1_) {

438

8

    new_stream->filter_manager_.streamInfo().setShouldDrainConnectionUponCompletion(true);

439

    // Prevent erroneous debug log of closing due to incoming connection close header.

440

8

    drain_state_ = DrainState::Closing;

441

8

442

443

94065

  new_stream->state_.is_internally_created_ = is_internally_created;

444

94065

  new_stream->response_encoder_ = &response_encoder;

445

94065

  new_stream->response_encoder_->getStream().addCallbacks(*new_stream);

446

94065

  new_stream->response_encoder_->getStream().registerCodecEventCallbacks(new_stream.get());

447

94065

  if (config_->streamFlushTimeout().has_value()) {

448

93285

    new_stream->response_encoder_->getStream().setFlushTimeout(

449

93285

        config_->streamFlushTimeout().value());

450

93942

  } else {

451

780

    new_stream->response_encoder_->getStream().setFlushTimeout(config_->streamIdleTimeout());

452

780

453

94065

  new_stream->streamInfo().setDownstreamBytesMeter(response_encoder.getStream().bytesMeter());

454

94065

  new_stream->streamInfo().setCodecStreamId(response_encoder.getStream().codecStreamId());

455

  // If the network connection is backed up, the stream should be made aware of it on creation.

456

  // Both HTTP/1.x and HTTP/2 codecs handle this in StreamCallbackHelper::addCallbacksHelper.

457

94065

  ASSERT(read_callbacks_->connection().aboveHighWatermark() == false ||

458

94065

         new_stream->filter_manager_.aboveHighWatermark());

459

94065

  LinkedList::moveIntoList(std::move(new_stream), streams_);

460

94065

  return **streams_.begin();

461

94065

462

463

void ConnectionManagerImpl::handleCodecErrorImpl(absl::string_view error, absl::string_view details,

464

1769

                                                 StreamInfo::CoreResponseFlag response_flag) {

465

1769

  ENVOY_CONN_LOG(debug, "dispatch error: {}", read_callbacks_->connection(), error);

466

1769

  read_callbacks_->connection().streamInfo().setResponseFlag(response_flag);

467

468

  // HTTP/1.1 codec has already sent a 400 response if possible. HTTP/2 codec has already sent

469

  // GOAWAY.

470

1769

  doConnectionClose(Network::ConnectionCloseType::FlushWriteAndDelay, response_flag, details);

471

1769

472

473

1759

void ConnectionManagerImpl::handleCodecError(absl::string_view error) {

474

1759

  handleCodecErrorImpl(error, absl::StrCat("codec_error:", StringUtil::replaceAllEmptySpace(error)),

475

1759

                       StreamInfo::CoreResponseFlag::DownstreamProtocolError);

476

1759

477

478

10

void ConnectionManagerImpl::handleCodecOverloadError(absl::string_view error) {

479

10

  handleCodecErrorImpl(error,

480

10

                       absl::StrCat("overload_error:", StringUtil::replaceAllEmptySpace(error)),

481

10

                       StreamInfo::CoreResponseFlag::OverloadManager);

482

10

483

484

22290

void ConnectionManagerImpl::createCodec(Buffer::Instance& data) {

485

22290

  ASSERT(!codec_);

486

22290

  codec_ = config_->createCodec(read_callbacks_->connection(), data, *this, overload_manager_);

487

488

22290

  switch (codec_->protocol()) {

489

1995

  case Protocol::Http3:

490

1995

    stats_.named_.downstream_cx_http3_total_.inc();

491

1995

    stats_.named_.downstream_cx_http3_active_.inc();

492

1995

    break;

493

7993

  case Protocol::Http2:

494

7993

    stats_.named_.downstream_cx_http2_total_.inc();

495

7993

    stats_.named_.downstream_cx_http2_active_.inc();

496

7993

    break;

497

12299

  case Protocol::Http11:

498

12302

  case Protocol::Http10:

499

12302

    stats_.named_.downstream_cx_http1_total_.inc();

500

12302

    stats_.named_.downstream_cx_http1_active_.inc();

501

12302

    break;

502

22290

503

22290

504

505

69564

Network::FilterStatus ConnectionManagerImpl::onData(Buffer::Instance& data, bool) {

506

69564

  requests_during_dispatch_count_ = 0;

507

69564

  if (!codec_) {

508

    // Close connections if Envoy is under pressure, typically memory, before creating codec.

509

20284

    if (hcm_ondata_creating_codec_ != nullptr && hcm_ondata_creating_codec_->shouldShedLoad()) {

510

4

      stats_.named_.downstream_rq_overload_close_.inc();

511

4

      handleCodecOverloadError("onData codec creation overload");

512

4

      return Network::FilterStatus::StopIteration;

513

4

514

    // Http3 codec should have been instantiated by now.

515

20280

    createCodec(data);

516

20280

517

518

69560

  bool redispatch;

519

69566

  do {

520

69566

    redispatch = false;

521

522

69566

    const Status status = codec_->dispatch(data);

523

524

69566

    if (isBufferFloodError(status) || isInboundFramesWithEmptyPayloadError(status)) {

525

33

      handleCodecError(status.message());

526

33

      return Network::FilterStatus::StopIteration;

527

69533

    } else if (isCodecProtocolError(status)) {

528

1711

      stats_.named_.downstream_cx_protocol_error_.inc();

529

1711

      handleCodecError(status.message());

530

1711

      return Network::FilterStatus::StopIteration;

531

67878

    } else if (isEnvoyOverloadError(status)) {

532

6

      stats_.named_.downstream_rq_overload_close_.inc();

533

6

      handleCodecOverloadError(status.message());

534

6

      return Network::FilterStatus::StopIteration;

535

6

536

67816

    ASSERT(status.ok());

537

538

    // Processing incoming data may release outbound data so check for closure here as well.

539

67816

    checkForDeferredClose(false);

540

541

    // The HTTP/1 codec will pause dispatch after a single message is complete. We want to

542

    // either redispatch if there are no streams and we have more data. If we have a single

543

    // complete non-WebSocket stream but have not responded yet we will pause socket reads

544

    // to apply back pressure.

545

67816

    if (codec_->protocol() < Protocol::Http2) {

546

17979

      if (read_callbacks_->connection().state() == Network::Connection::State::Open &&

547

17979

          data.length() > 0 && streams_.empty()) {

548

6

        redispatch = true;

549

6

550

17979

551

67816

  } while (redispatch);

552

553

67810

  if (!read_callbacks_->connection().streamInfo().protocol()) {

554

18615

    read_callbacks_->connection().streamInfo().protocol(codec_->protocol());

555

18615

556

557

67810

  return Network::FilterStatus::StopIteration;

558

69560

559

560

22185

Network::FilterStatus ConnectionManagerImpl::onNewConnection() {

561

22185

  if (!read_callbacks_->connection().streamInfo().protocol()) {

562

    // For Non-QUIC traffic, continue passing data to filters.

563

20191

    return Network::FilterStatus::Continue;

564

20191

565

  // Only QUIC connection's stream_info_ specifies protocol.

566

1994

  Buffer::OwnedImpl dummy;

567

1994

  createCodec(dummy);

568

1994

  ASSERT(codec_->protocol() == Protocol::Http3);

569

  // Stop iterating through network filters for QUIC. Currently QUIC connections bypass the

570

  // onData() interface because QUICHE already handles de-multiplexing.

571

1994

  return Network::FilterStatus::StopIteration;

572

22185

573

574

void ConnectionManagerImpl::resetAllStreams(

575

667

    absl::optional<StreamInfo::CoreResponseFlag> response_flag, absl::string_view details) {

576

45397

  while (!streams_.empty()) {

577

    // Mimic a downstream reset in this case. We must also remove callbacks here. Though we are

578

    // about to close the connection and will disable further reads, it is possible that flushing

579

    // data out can cause stream callbacks to fire (e.g., low watermark callbacks).

580

//

581

    // TODO(mattklein123): I tried to actually reset through the codec here, but ran into issues

582

    // with nghttp2 state and being unhappy about sending reset frames after the connection had

583

    // been terminated via GOAWAY. It might be possible to do something better here inside the h2

584

    // codec but there are no easy answers and this seems simpler.

585

44730

    auto& stream = *streams_.front();

586

44730

    stream.response_encoder_->getStream().removeCallbacks(stream);

587

44730

    if (!stream.response_encoder_->getStream().responseDetails().empty()) {

588

323

      stream.filter_manager_.streamInfo().setResponseCodeDetails(

589

323

          stream.response_encoder_->getStream().responseDetails());

590

44695

    } else if (!details.empty()) {

591

44407

      stream.filter_manager_.streamInfo().setResponseCodeDetails(details);

592

44407

593

44730

    if (response_flag.has_value()) {

594

41113

      stream.filter_manager_.streamInfo().setResponseFlag(response_flag.value());

595

41113

596

44730

    stream.onResetStream(StreamResetReason::ConnectionTermination, absl::string_view());

597

44730

598

667

599

600

44039

void ConnectionManagerImpl::onEvent(Network::ConnectionEvent event) {

601

44039

  if (event == Network::ConnectionEvent::LocalClose) {

602

3124

    stats_.named_.downstream_cx_destroy_local_.inc();

603

3124

604

605

44039

  if (event == Network::ConnectionEvent::RemoteClose ||

606

44039

      event == Network::ConnectionEvent::LocalClose) {

607

608

22353

    std::string details;

609

22353

    if (event == Network::ConnectionEvent::RemoteClose) {

610

19229

      remote_close_ = true;

611

19229

      stats_.named_.downstream_cx_destroy_remote_.inc();

612

19229

      details = StreamInfo::ResponseCodeDetails::get().DownstreamRemoteDisconnect;

613

19522

    } else {

614

3124

      absl::string_view local_close_reason = read_callbacks_->connection().localCloseReason();

615

3124

      ENVOY_BUG(!local_close_reason.empty(), "Local Close Reason was not set!");

616

3124

      details = fmt::format(

617

3124

          fmt::runtime(StreamInfo::ResponseCodeDetails::get().DownstreamLocalDisconnect),

618

3124

          StringUtil::replaceAllEmptySpace(local_close_reason));

619

3124

620

621

    // TODO(mattklein123): It is technically possible that something outside of the filter causes

622

    // a local connection close, so we still guard against that here. A better solution would be to

623

    // have some type of "pre-close" callback that we could hook for cleanup that would get called

624

    // regardless of where local close is invoked from.

625

    // NOTE: that this will cause doConnectionClose() to get called twice in the common local close

626

    // cases, but the method protects against that.

627

    // NOTE: In the case where a local close comes from outside the filter, this will cause any

628

    // stream closures to increment remote close stats. We should do better here in the future,

629

    // via the pre-close callback mentioned above.

630

22353

    doConnectionClose(absl::nullopt, StreamInfo::CoreResponseFlag::DownstreamConnectionTermination,

631

22353

                      details);

632

22353

633

44039

634

635

void ConnectionManagerImpl::doConnectionClose(

636

    absl::optional<Network::ConnectionCloseType> close_type,

637

26613

    absl::optional<StreamInfo::CoreResponseFlag> response_flag, absl::string_view details) {

638

26613

  if (connection_idle_timer_) {

639

21756

    connection_idle_timer_->disableTimer();

640

21756

    connection_idle_timer_.reset();

641

21756

642

643

26613

  if (connection_duration_timer_) {

644

85

    connection_duration_timer_->disableTimer();

645

85

    connection_duration_timer_.reset();

646

85

647

648

26613

  if (drain_timer_) {

649

187

    drain_timer_->disableTimer();

650

187

    drain_timer_.reset();

651

187

652

653

26613

  if (!streams_.empty()) {

654

667

    const Network::ConnectionEvent event = close_type.has_value()

655

667

                                               ? Network::ConnectionEvent::LocalClose

656

667

                                               : Network::ConnectionEvent::RemoteClose;

657

667

    if (event == Network::ConnectionEvent::LocalClose) {

658

377

      stats_.named_.downstream_cx_destroy_local_active_rq_.inc();

659

377

660

667

    if (event == Network::ConnectionEvent::RemoteClose) {

661

290

      stats_.named_.downstream_cx_destroy_remote_active_rq_.inc();

662

290

663

664

667

    stats_.named_.downstream_cx_destroy_active_rq_.inc();

665

667

    user_agent_.onConnectionDestroy(event, true);

666

    // Note that resetAllStreams() does not actually write anything to the wire. It just resets

667

    // all upstream streams and their filter stacks. Thus, there are no issues around recursive

668

    // entry.

669

667

    resetAllStreams(response_flag, details);

670

667

671

672

26613

  if (close_type.has_value()) {

673

4260

    read_callbacks_->connection().close(close_type.value(), details);

674

4260

675

26613

676

677

96765

bool ConnectionManagerImpl::isPrematureRstStream(const ActiveStream& stream) const {

678

  // Check if the request was prematurely reset, by comparing its lifetime to the configured

679

  // threshold.

680

96765

  ASSERT(!stream.state_.is_internally_destroyed_);

681

96765

  absl::optional<std::chrono::nanoseconds> duration =

682

96765

      stream.filter_manager_.streamInfo().currentDuration();

683

684

  // Check if request lifetime is longer than the premature reset threshold.

685

96765

  if (duration) {

686

96765

    const uint64_t lifetime = std::chrono::duration_cast<std::chrono::seconds>(*duration).count();

687

96765

    const uint64_t min_lifetime = runtime_.snapshot().getInteger(

688

96765

        ConnectionManagerImpl::PrematureResetMinStreamLifetimeSecondsKey, 1);

689

96765

    if (lifetime > min_lifetime) {

690

33820

      return false;

691

33820

692

96765

693

694

  // If request has completed before configured threshold, also check if the Envoy proxied the

695

  // response from the upstream. Requests without the response status were reset.

696

  // TODO(RyanTheOptimist): Possibly support half_closed_local instead.

697

62945

  return !stream.filter_manager_.streamInfo().responseCode();

698

96765

699

700

// Sends a GOAWAY if too many streams have been reset prematurely on this

701

// connection.

702

94065

void ConnectionManagerImpl::maybeDrainDueToPrematureResets() {

703

  // If the connection has been drained due to premature resets, do not check this again.

704

  // Without this flag, recursion may occur, as shown in the following stack trace:

705

//

706

  //   maybeDrainDueToPrematureResets()

707

  //   doConnectionClose()

708

  //   resetAllStreams()

709

  //   onResetStream()

710

  //   doDeferredStreamDestroy()

711

  //   maybeDrainDueToPrematureResets()

712

  //   ...

713

//

714

  // The recursion will continue until all streams are destroyed. If there are many streams

715

  // that may result in a stack overflow. This flag is used to avoid above recursion.

716

94065

  if (drained_due_to_premature_resets_) {

717

3600

    return;

718

3600

719

720

90465

  if (closed_non_internally_destroyed_requests_ == 0) {

721

266

    return;

722

266

723

724

90199

  const uint64_t limit =

725

90199

      runtime_.snapshot().getInteger(ConnectionManagerImpl::PrematureResetTotalStreamCountKey, 500);

726

727

90199

  if (closed_non_internally_destroyed_requests_ < limit) {

728

    // Even though the total number of streams have not reached `limit`, check if the number of bad

729

    // streams is high enough that even if every subsequent stream is good, the connection

730

    // would be closed once the limit is reached, and if so close the connection now.

731

37286

    if (number_premature_stream_resets_ * 2 < limit) {

732

36280

      return;

733

36280

734

82492

  } else {

735

52913

    if (number_premature_stream_resets_ * 2 < closed_non_internally_destroyed_requests_) {

736

20965

      return;

737

20965

738

52913

739

740

32954

  if (read_callbacks_->connection().state() == Network::Connection::State::Open) {

741

8

    stats_.named_.downstream_rq_too_many_premature_resets_.inc();

742

743

    // Mark the the connection has been drained due to too many premature resets.

744

8

    drained_due_to_premature_resets_ = true;

745

746

8

    doConnectionClose(Network::ConnectionCloseType::Abort, absl::nullopt,

747

8

                      "too_many_premature_resets");

748

8

749

32954

750

751

1

void ConnectionManagerImpl::onGoAway(GoAwayErrorCode) {

752

  // Currently we do nothing with remote go away frames. In the future we can decide to no longer

753

  // push resources if applicable.

754

1

755

756

38

void ConnectionManagerImpl::onIdleTimeout() {

757

38

  ENVOY_CONN_LOG(debug, "idle timeout", read_callbacks_->connection());

758

38

  stats_.named_.downstream_cx_idle_timeout_.inc();

759

38

  if (!codec_) {

760

    // No need to delay close after flushing since an idle timeout has already fired. Attempt to

761

    // write out buffered data one last time and issue a local close if successful.

762

1

    doConnectionClose(Network::ConnectionCloseType::FlushWrite, absl::nullopt,

763

1

                      StreamInfo::LocalCloseReasons::get().IdleTimeoutOnConnection);

764

38

  } else if (drain_state_ == DrainState::NotDraining) {

765

37

    startDrainSequence();

766

37

767

38

768

769

85

void ConnectionManagerImpl::onConnectionDurationTimeout() {

770

85

  ENVOY_CONN_LOG(debug, "max connection duration reached", read_callbacks_->connection());

771

85

  stats_.named_.downstream_cx_max_duration_reached_.inc();

772

85

  if (!codec_) {

773

    // Attempt to write out buffered data one last time and issue a local close if successful.

774

8

    doConnectionClose(Network::ConnectionCloseType::FlushWrite,

775

8

                      StreamInfo::CoreResponseFlag::DurationTimeout,

776

8

                      StreamInfo::ResponseCodeDetails::get().DurationTimeout);

777

78

  } else if (drain_state_ == DrainState::NotDraining) {

778

77

    if (config_->http1SafeMaxConnectionDuration() && codec_->protocol() < Protocol::Http2) {

779

9

      ENVOY_CONN_LOG(debug,

780

9

                     "HTTP1-safe max connection duration is configured -- skipping drain sequence.",

781

9

                     read_callbacks_->connection());

782

9

      stats_.named_.downstream_cx_http1_soft_drain_.inc();

783

9

      soft_drain_http1_ = true;

784

70

    } else {

785

68

      startDrainSequence();

786

68

787

77

788

85

789

790

124

void ConnectionManagerImpl::onDrainTimeout() {

791

124

  ASSERT(drain_state_ != DrainState::NotDraining);

792

124

  codec_->goAway();

793

124

  drain_state_ = DrainState::Closing;

794

124

  checkForDeferredClose(false);

795

124

796

797

18

void ConnectionManagerImpl::sendGoAwayAndClose(bool graceful) {

798

18

  ENVOY_CONN_LOG(trace, "connection manager sendGoAwayAndClose was triggerred from filters.",

799

18

                 read_callbacks_->connection());

800

18

  if (go_away_sent_) {

801

    return;

802

803

804

  // Use graceful drain sequence if graceful shutdown is requested.

805

  // startDrainSequence() works for both HTTP/1 and HTTP/2:

806

  // - HTTP/1: shutdownNotice() + goAway() provides graceful close

807

  // - HTTP/2: shutdownNotice() sends GOAWAY with high stream ID, then goAway() sends final GOAWAY

808

18

  if (graceful) {

809

1

    if (drain_state_ == DrainState::NotDraining) {

810

1

      startDrainSequence();

811

1

812

    // Consider the "go away" process started once draining begins.

813

    // The actual GOAWAY frame will be sent in onDrainTimeout(), but we want to

814

    // prevent multiple calls to sendGoAwayAndClose() from starting multiple drain sequences.

815

1

    go_away_sent_ = true;

816

18

  } else {

817

    // Immediate close - send GOAWAY and close immediately

818

17

    codec_->shutdownNotice();

819

17

    codec_->goAway();

820

17

    go_away_sent_ = true;

821

17

    doConnectionClose(Network::ConnectionCloseType::FlushWriteAndDelay, absl::nullopt,

822

17

                      "forced_goaway");

823

17

824

18

825

826

void ConnectionManagerImpl::chargeTracingStats(const Tracing::Reason& tracing_reason,

827

271

                                               ConnectionManagerTracingStats& tracing_stats) {

828

271

  switch (tracing_reason) {

829

1

  case Tracing::Reason::ClientForced:

830

1

    tracing_stats.client_enabled_.inc();

831

1

    break;

832

40

  case Tracing::Reason::Sampling:

833

40

    tracing_stats.random_sampling_.inc();

834

40

    break;

835

11

  case Tracing::Reason::ServiceForced:

836

11

    tracing_stats.service_forced_.inc();

837

11

    break;

838

219

  default:

839

219

    tracing_stats.not_traceable_.inc();

840

219

    break;

841

271

842

271

843

844

absl::optional<absl::string_view>

845

3031

ConnectionManagerImpl::HttpStreamIdProviderImpl::toStringView() const {

846

3031

  if (parent_.request_headers_ == nullptr) {

847

1

    return {};

848

1

849

3030

  ASSERT(parent_.connection_manager_.config_->requestIDExtension() != nullptr);

850

3030

  return parent_.connection_manager_.config_->requestIDExtension()->get(*parent_.request_headers_);

851

3031

852

853

1

absl::optional<uint64_t> ConnectionManagerImpl::HttpStreamIdProviderImpl::toInteger() const {

854

1

  if (parent_.request_headers_ == nullptr) {

855

    return {};

856

857

1

  ASSERT(parent_.connection_manager_.config_->requestIDExtension() != nullptr);

858

1

  return parent_.connection_manager_.config_->requestIDExtension()->getInteger(

859

1

      *parent_.request_headers_);

860

1

861

862

namespace {

863

constexpr absl::string_view kRouteFactoryName = "envoy.route_config_update_requester.default";

864

} // namespace

865

866

ConnectionManagerImpl::ActiveStream::ActiveStream(ConnectionManagerImpl& connection_manager,

867

                                                  uint32_t buffer_limit,

868

                                                  Buffer::BufferMemoryAccountSharedPtr account)

869

94065

    : connection_manager_(connection_manager),

870

94065

      connection_manager_tracing_config_(connection_manager_.config_->tracingConfig() == nullptr

871

94065

                                             ? absl::nullopt

872

94065

                                             : makeOptRef<const TracingConnectionManagerConfig>(

873

346

                                                   *connection_manager_.config_->tracingConfig())),

874

94065

      stream_id_(connection_manager.random_generator_.random()),

875

94065

      filter_manager_(*this, *connection_manager_.dispatcher_,

876

94065

                      connection_manager_.read_callbacks_->connection(), stream_id_,

877

94065

                      std::move(account), connection_manager_.config_->proxy100Continue(),

878

94065

                      buffer_limit, connection_manager_.config_->filterFactory(),

879

94065

                      connection_manager_.config_->localReply(),

880

94065

                      connection_manager_.codec_->protocol(), connection_manager_.timeSource(),

881

94065

                      connection_manager_.read_callbacks_->connection().streamInfo().filterState(),

882

94065

                      connection_manager_.overload_manager_),

883

94065

      request_response_timespan_(new Stats::HistogramCompletableTimespanImpl(

884

94065

          connection_manager_.stats_.named_.downstream_rq_time_, connection_manager_.timeSource())),

885

      has_explicit_global_flush_timeout_(

886

94065

          connection_manager.config_->streamFlushTimeout().has_value()),

887

      header_validator_(

888

94065

          connection_manager.config_->makeHeaderValidator(connection_manager.codec_->protocol())),

889

94065

      trace_refresh_after_route_refresh_(Runtime::runtimeFeatureEnabled(

890

94065

          "envoy.reloadable_features.trace_refresh_after_route_refresh")) {

891

94065

  ASSERT(!connection_manager.config_->isRoutable() ||

892

94065

             ((connection_manager.config_->routeConfigProvider() == nullptr &&

893

94065

               connection_manager.config_->scopedRouteConfigProvider() != nullptr &&

894

94065

               connection_manager.config_->scopeKeyBuilder().has_value()) ||

895

94065

              (connection_manager.config_->routeConfigProvider() != nullptr &&

896

94065

               connection_manager.config_->scopedRouteConfigProvider() == nullptr &&

897

94065

               !connection_manager.config_->scopeKeyBuilder().has_value())),

898

94065

         "Either routeConfigProvider or (scopedRouteConfigProvider and scopeKeyBuilder) should be "

899

94065

         "set in "

900

94065

         "ConnectionManagerImpl.");

901

902

94065

  filter_manager_.streamInfo().setStreamIdProvider(

903

94065

      std::make_shared<HttpStreamIdProviderImpl>(*this));

904

905

94065

  filter_manager_.streamInfo().setShouldSchemeMatchUpstream(

906

94065

      connection_manager.config_->shouldSchemeMatchUpstream());

907

908

  // TODO(chaoqin-li1123): can this be moved to the on demand filter?

909

94065

  auto factory = Envoy::Config::Utility::getFactoryByName<RouteConfigUpdateRequesterFactory>(

910

94065

      kRouteFactoryName);

911

94065

  if (connection_manager_.config_->isRoutable() &&

912

94065

      connection_manager.config_->routeConfigProvider() != nullptr && factory) {

913

93441

    route_config_update_requester_ = factory->createRouteConfigUpdateRequester(

914

93441

        connection_manager.config_->routeConfigProvider());

915

93848

  } else if (connection_manager_.config_->isRoutable() &&

916

624

             connection_manager.config_->scopedRouteConfigProvider() != nullptr &&

917

624

             connection_manager.config_->scopeKeyBuilder().has_value() && factory) {

918

183

    route_config_update_requester_ = factory->createRouteConfigUpdateRequester(

919

183

        connection_manager.config_->scopedRouteConfigProvider(),

920

183

        connection_manager.config_->scopeKeyBuilder());

921

183

922

94065

  ScopeTrackerScopeState scope(this,

923

94065

                               connection_manager_.read_callbacks_->connection().dispatcher());

924

925

94065

  connection_manager_.stats_.named_.downstream_rq_total_.inc();

926

94065

  connection_manager_.stats_.named_.downstream_rq_active_.inc();

927

94065

  if (connection_manager_.codec_->protocol() == Protocol::Http2) {

928

74740

    connection_manager_.stats_.named_.downstream_rq_http2_total_.inc();

929

88741

  } else if (connection_manager_.codec_->protocol() == Protocol::Http3) {

930

2529

    connection_manager_.stats_.named_.downstream_rq_http3_total_.inc();

931

17995

  } else {

932

16796

    connection_manager_.stats_.named_.downstream_rq_http1_total_.inc();

933

16796

934

935

94065

  if (connection_manager_.config_->streamIdleTimeout().count()) {

936

93293

    idle_timeout_ms_ = connection_manager_.config_->streamIdleTimeout();

937

93293

    stream_idle_timer_ = connection_manager_.dispatcher_->createScaledTimer(

938

93293

        Event::ScaledTimerType::HttpDownstreamIdleStreamTimeout,

939

93293

        [this]() -> void { onIdleTimeout(); });

940

93293

    resetIdleTimer();

941

93293

942

943

94065

  if (connection_manager_.config_->requestTimeout().count()) {

944

71

    std::chrono::milliseconds request_timeout = connection_manager_.config_->requestTimeout();

945

71

    request_timer_ =

946

71

        connection_manager.dispatcher_->createTimer([this]() -> void { onRequestTimeout(); });

947

71

    request_timer_->enableTimer(request_timeout, this);

948

71

949

950

94065

  if (connection_manager_.config_->requestHeadersTimeout().count()) {

951

2

    std::chrono::milliseconds request_headers_timeout =

952

2

        connection_manager_.config_->requestHeadersTimeout();

953

2

    request_header_timer_ =

954

2

        connection_manager.dispatcher_->createTimer([this]() -> void { onRequestHeaderTimeout(); });

955

2

    request_header_timer_->enableTimer(request_headers_timeout, this);

956

2

957

958

94065

  const auto max_stream_duration = connection_manager_.config_->maxStreamDuration();

959

94065

  if (max_stream_duration.has_value() && max_stream_duration.value().count()) {

960

38

    max_stream_duration_timer_ = connection_manager.dispatcher_->createTimer(

961

38

        [this]() -> void { onStreamMaxDurationReached(); });

962

38

    max_stream_duration_timer_->enableTimer(

963

38

        connection_manager_.config_->maxStreamDuration().value(), this);

964

38

965

966

94065

  if (connection_manager_.config_->accessLogFlushInterval().has_value()) {

967

11

    access_log_flush_timer_ = connection_manager.dispatcher_->createTimer([this]() -> void {

968

      // If the request is complete, we've already done the stream-end access-log, and shouldn't

969

      // do the periodic log.

970

11

      if (!streamInfo().requestComplete().has_value()) {

971

11

        log(AccessLog::AccessLogType::DownstreamPeriodic);

972

11

        refreshAccessLogFlushTimer();

973

11

974

11

      const SystemTime now = connection_manager_.timeSource().systemTime();

975

      // Downstream bytes meter is guaranteed to be non-null because ActiveStream and the timer

976

      // event are created on the same thread that sets the meter in

977

      // ConnectionManagerImpl::newStream.

978

11

      filter_manager_.streamInfo().getDownstreamBytesMeter()->takeDownstreamPeriodicLoggingSnapshot(

979

11

          now);

980

11

      if (auto& upstream_bytes_meter = filter_manager_.streamInfo().getUpstreamBytesMeter();

981

11

          upstream_bytes_meter != nullptr) {

982

11

        upstream_bytes_meter->takeDownstreamPeriodicLoggingSnapshot(now);

983

11

984

11

});

985

10

    refreshAccessLogFlushTimer();

986

10

987

94065

988

989

92585

void ConnectionManagerImpl::ActiveStream::log(AccessLog::AccessLogType type) {

990

92585

  const Formatter::Context log_context{

991

92585

      request_headers_.get(), response_headers_.get(), response_trailers_.get(), {}, type,

992

92585

      active_span_.get()};

993

994

92585

  filter_manager_.log(log_context);

995

996

92758

  for (const auto& access_logger : connection_manager_.config_->accessLogs()) {

997

72951

    access_logger->log(log_context, filter_manager_.streamInfo());

998

72951

999

92585

1000

1001

94065

void ConnectionManagerImpl::ActiveStream::completeRequest() {

1002

94065

  filter_manager_.streamInfo().onRequestComplete();

1003

1004

94065

  connection_manager_.stats_.named_.downstream_rq_active_.dec();

1005

94065

  if (filter_manager_.streamInfo().healthCheck()) {

1006

264

    connection_manager_.config_->tracingStats().health_check_.inc();

1007

264

1008

1009

94065

  if (active_span_) {

1010

52

    Tracing::HttpTracerUtility::finalizeDownstreamSpan(

1011

52

        *active_span_, request_headers_.get(), response_headers_.get(), response_trailers_.get(),

1012

52

        filter_manager_.streamInfo(), *this);

1013

52

1014

94065

  if (state_.successful_upgrade_) {

1015

601

    connection_manager_.stats_.named_.downstream_cx_upgrades_active_.dec();

1016

601

1017

94065

1018

1019

564933

void ConnectionManagerImpl::ActiveStream::resetIdleTimer() {

1020

564933

  if (stream_idle_timer_ != nullptr) {

1021

    // TODO(htuch): If this shows up in performance profiles, optimize by only

1022

    // updating a timestamp here and doing periodic checks for idle timeouts

1023

    // instead, or reducing the accuracy of timers.

1024

557464

    stream_idle_timer_->enableTimer(idle_timeout_ms_);

1025

557464

1026

564933

1027

1028

184

void ConnectionManagerImpl::ActiveStream::onIdleTimeout() {

1029

184

  connection_manager_.stats_.named_.downstream_rq_idle_timeout_.inc();

1030

1031

184

  filter_manager_.streamInfo().setResponseFlag(StreamInfo::CoreResponseFlag::StreamIdleTimeout);

1032

184

  sendLocalReply(

1033

184

      Http::Utility::maybeRequestTimeoutCode(filter_manager_.hasLastDownstreamByteReceived()),

1034

184

      "stream timeout", nullptr, absl::nullopt,

1035

184

      StreamInfo::ResponseCodeDetails::get().StreamIdleTimeout);

1036

184

1037

1038

27

void ConnectionManagerImpl::ActiveStream::onRequestTimeout() {

1039

27

  connection_manager_.stats_.named_.downstream_rq_timeout_.inc();

1040

27

  sendLocalReply(

1041

27

      Http::Utility::maybeRequestTimeoutCode(filter_manager_.hasLastDownstreamByteReceived()),

1042

27

      "request timeout", nullptr, absl::nullopt,

1043

27

      StreamInfo::ResponseCodeDetails::get().RequestOverallTimeout);

1044

27

1045

1046

1

void ConnectionManagerImpl::ActiveStream::onRequestHeaderTimeout() {

1047

1

  connection_manager_.stats_.named_.downstream_rq_header_timeout_.inc();

1048

1

  sendLocalReply(

1049

1

      Http::Utility::maybeRequestTimeoutCode(filter_manager_.hasLastDownstreamByteReceived()),

1050

1

      "request header timeout", nullptr, absl::nullopt,

1051

1

      StreamInfo::ResponseCodeDetails::get().RequestHeaderTimeout);

1052

1

1053

1054

32

void ConnectionManagerImpl::ActiveStream::onStreamMaxDurationReached() {

1055

32

  ENVOY_STREAM_LOG(debug, "Stream max duration time reached", *this);

1056

32

  connection_manager_.stats_.named_.downstream_rq_max_duration_reached_.inc();

1057

32

  sendLocalReply(

1058

32

      Http::Utility::maybeRequestTimeoutCode(filter_manager_.hasLastDownstreamByteReceived()),

1059

32

      "downstream duration timeout", nullptr, Grpc::Status::WellKnownGrpcStatus::DeadlineExceeded,

1060

32

      StreamInfo::ResponseCodeDetails::get().MaxDurationTimeout);

1061

32

1062

1063

47577

void ConnectionManagerImpl::ActiveStream::chargeStats(const ResponseHeaderMap& headers) {

1064

47577

  if (trace_refresh_after_route_refresh_ && connection_manager_tracing_config_.has_value()) {

1065

266

    const Tracing::Decision tracing_decision =

1066

266

        Tracing::TracerUtility::shouldTraceRequest(filter_manager_.streamInfo());

1067

266

    ConnectionManagerImpl::chargeTracingStats(tracing_decision.reason,

1068

266

                                              connection_manager_.config_->tracingStats());

1069

266

1070

1071

47577

  uint64_t response_code = Utility::getResponseStatus(headers);

1072

47577

  filter_manager_.streamInfo().setResponseCode(response_code);

1073

1074

47577

  if (filter_manager_.streamInfo().healthCheck()) {

1075

264

    return;

1076

264

1077

1078

  // No response is sent back downstream for internal redirects, so don't charge downstream stats.

1079

47313

  const absl::optional<std::string>& response_code_details =

1080

47313

      filter_manager_.streamInfo().responseCodeDetails();

1081

47313

  if (response_code_details.has_value() &&

1082

47313

      response_code_details == Envoy::StreamInfo::ResponseCodeDetails::get().InternalRedirect) {

1083

151

    return;

1084

151

1085

1086

47162

  connection_manager_.stats_.named_.downstream_rq_completed_.inc();

1087

47162

  connection_manager_.listener_stats_.downstream_rq_completed_.inc();

1088

47162

  if (CodeUtility::is1xx(response_code)) {

1089

253

    connection_manager_.stats_.named_.downstream_rq_1xx_.inc();

1090

253

    connection_manager_.listener_stats_.downstream_rq_1xx_.inc();

1091

46917

  } else if (CodeUtility::is2xx(response_code)) {

1092

37231

    connection_manager_.stats_.named_.downstream_rq_2xx_.inc();

1093

37231

    connection_manager_.listener_stats_.downstream_rq_2xx_.inc();

1094

41922

  } else if (CodeUtility::is3xx(response_code)) {

1095

194

    connection_manager_.stats_.named_.downstream_rq_3xx_.inc();

1096

194

    connection_manager_.listener_stats_.downstream_rq_3xx_.inc();

1097

9584

  } else if (CodeUtility::is4xx(response_code)) {

1098

5726

    connection_manager_.stats_.named_.downstream_rq_4xx_.inc();

1099

5726

    connection_manager_.listener_stats_.downstream_rq_4xx_.inc();

1100

8947

  } else if (CodeUtility::is5xx(response_code)) {

1101

3746

    connection_manager_.stats_.named_.downstream_rq_5xx_.inc();

1102

3746

    connection_manager_.listener_stats_.downstream_rq_5xx_.inc();

1103

3746

1104

47162

1105

1106

91795

const Network::Connection* ConnectionManagerImpl::ActiveStream::connection() {

1107

91795

  return &connection_manager_.read_callbacks_->connection();

1108

91795

1109

1110

90323

uint32_t ConnectionManagerImpl::ActiveStream::localPort() {

1111

90323

  auto ip = connection()->connectionInfoProvider().localAddress()->ip();

1112

90323

  if (ip == nullptr) {

1113

15

    return 0;

1114

15

1115

90308

  return ip->port();

1116

90323

1117

1118

namespace {

1119

5

bool streamErrorOnlyErrors(absl::string_view error_details) {

1120

  // Pre UHV HCM did not respect stream_error_on_invalid_http_message

1121

  // and only sent 400 for specific errors.

1122

  // TODO(#28555): make these errors respect the stream_error_on_invalid_http_message

1123

5

  return error_details == UhvResponseCodeDetail::get().FragmentInUrlPath ||

1124

5

         error_details == UhvResponseCodeDetail::get().EscapedSlashesInPath ||

1125

5

         error_details == UhvResponseCodeDetail::get().Percent00InPath;

1126

5

1127

} // namespace

1128

1129

51

void ConnectionManagerImpl::ActiveStream::setRequestDecorator(RequestHeaderMap& headers) {

1130

51

  ASSERT(active_span_ != nullptr);

1131

51

  const Router::Decorator* decorater = route_decorator_;

1132

1133

  // If a decorator has been defined, apply it to the active span.

1134

51

  absl::string_view decorated_operation;

1135

51

  if (decorater != nullptr) {

1136

11

    decorated_operation = decorater->getOperation();

1137

1138

11

    decorater->apply(*active_span_);

1139

11

    state_.decorated_propagate_ = decorater->propagate();

1140

11

1141

1142

51

  if (connection_manager_tracing_config_->operation_name_ == Tracing::OperationName::Egress) {

1143

    // For egress (outbound) requests, pass the decorator's operation name (if defined and

1144

    // propagation enabled) as a request header to enable the receiving service to use it in its

1145

    // server span.

1146

3

    if (!decorated_operation.empty() && state_.decorated_propagate_) {

1147

2

      headers.setEnvoyDecoratorOperation(decorated_operation);

1148

2

1149

48

  } else {

1150

48

    absl::string_view req_operation_override = headers.getEnvoyDecoratorOperationValue();

1151

1152

    // For ingress (inbound) requests, if a decorator operation name has been provided, it

1153

    // should be used to override the active span's operation.

1154

48

    if (!req_operation_override.empty()) {

1155

1

      active_span_->setOperation(req_operation_override);

1156

1157

      // Set the decorator operation as overridden to avoid propagating the route decorator

1158

      // operation to the client when the setResponseDecorator() is called.

1159

1

      state_.decorator_overriden_ = true;

1160

1

1161

    // Remove header so not propagated to service

1162

48

    headers.removeEnvoyDecoratorOperation();

1163

48

1164

51

1165

1166

49

void ConnectionManagerImpl::ActiveStream::setResponseDecorator(ResponseHeaderMap& headers) {

1167

49

  ASSERT(active_span_ != nullptr);

1168

1169

49

  if (connection_manager_tracing_config_->operation_name_ == Tracing::OperationName::Ingress) {

1170

    // For ingress (inbound) responses, if the request headers do not include a

1171

    // decorator operation (override), and the decorated operation should be

1172

    // propagated, then pass the decorator's operation name (if defined)

1173

    // as a response header to enable the client service to use it in its client span.

1174

46

    if (state_.decorated_propagate_ && !state_.decorator_overriden_) {

1175

44

      absl::string_view decorated_operation =

1176

44

          route_decorator_ != nullptr ? route_decorator_->getOperation() : absl::string_view();

1177

1178

44

      if (!decorated_operation.empty()) {

1179

        // If the decorator operation is defined, set it as the response header.

1180

4

        headers.setEnvoyDecoratorOperation(decorated_operation);

1181

4

1182

44

1183

46

  } else if (connection_manager_tracing_config_->operation_name_ ==

1184

3

             Tracing::OperationName::Egress) {

1185

3

    const absl::string_view resp_operation_override = headers.getEnvoyDecoratorOperationValue();

1186

1187

    // For Egress (outbound) response, if a decorator operation name has been provided, it

1188

    // should be used to override the active span's operation.

1189

3

    if (!resp_operation_override.empty()) {

1190

1

      active_span_->setOperation(resp_operation_override);

1191

1

1192

    // Remove header so not propagated to service.

1193

3

    headers.removeEnvoyDecoratorOperation();

1194

3

1195

49

1196

1197

91472

bool ConnectionManagerImpl::ActiveStream::validateHeaders() {

1198

91472

  if (header_validator_) {

1199

10

    auto validation_result = header_validator_->validateRequestHeaders(*request_headers_);

1200

10

    bool failure = !validation_result.ok();

1201

10

    bool redirect = false;

1202

10

    bool is_grpc = Grpc::Common::hasGrpcContentType(*request_headers_);

1203

10

    std::string failure_details(validation_result.details());

1204

10

    if (!failure) {

1205

7

      auto transformation_result = header_validator_->transformRequestHeaders(*request_headers_);

1206

7

      failure = !transformation_result.ok();

1207

7

      redirect = transformation_result.action() ==

1208

7

                 Http::ServerHeaderValidator::RequestHeadersTransformationResult::Action::Redirect;

1209

7

      failure_details = std::string(transformation_result.details());

1210

7

      if (redirect && !is_grpc) {

1211

1

        connection_manager_.stats_.named_.downstream_rq_redirected_with_normalized_path_.inc();

1212

6

      } else if (failure) {

1213

1

        connection_manager_.stats_.named_.downstream_rq_failed_path_normalization_.inc();

1214

1

1215

7

1216

10

    if (failure) {

1217

5

      std::function<void(ResponseHeaderMap & headers)> modify_headers;

1218

5

      Code response_code = failure_details == Http1ResponseCodeDetail::get().InvalidTransferEncoding

1219

5

                               ? Code::NotImplemented

1220

5

                               : Code::BadRequest;

1221

5

      absl::optional<Grpc::Status::GrpcStatus> grpc_status;

1222

5

      if (redirect && !is_grpc) {

1223

1

        response_code = Code::TemporaryRedirect;

1224

1

        modify_headers = [new_path = request_headers_->Path()->value().getStringView()](

1225

1

                             Http::ResponseHeaderMap& response_headers) -> void {

1226

1

          response_headers.addReferenceKey(Http::Headers::get().Location, new_path);

1227

1

};

1228

5

      } else if (is_grpc) {

1229

2

        grpc_status = Grpc::Status::WellKnownGrpcStatus::Internal;

1230

2

1231

1232

5

      filter_manager_.streamInfo().setResponseFlag(

1233

5

          StreamInfo::CoreResponseFlag::DownstreamProtocolError);

1234

1235

      // H/2 codec was resetting requests that were rejected due to headers with underscores,

1236

      // instead of sending 400. Preserving this behavior for now.

1237

      // TODO(#24466): Make H/2 behavior consistent with H/1 and H/3.

1238

5

      if (failure_details == UhvResponseCodeDetail::get().InvalidUnderscore &&

1239

5

          connection_manager_.codec_->protocol() == Protocol::Http2) {

1240

        filter_manager_.streamInfo().setResponseCodeDetails(failure_details);

1241

        resetStream();

1242

5

      } else {

1243

5

        sendLocalReply(response_code, "", modify_headers, grpc_status, failure_details);

1244

5

        if (!response_encoder_->streamErrorOnInvalidHttpMessage() &&

1245

5

            !streamErrorOnlyErrors(failure_details)) {

1246

5

          connection_manager_.handleCodecError(failure_details);

1247

5

1248

5

1249

5

      return false;

1250

5

1251

10

1252

1253

91467

  return true;

1254

91472

1255

1256

493

bool ConnectionManagerImpl::ActiveStream::validateTrailers(RequestTrailerMap& trailers) {

1257

493

  if (!header_validator_) {

1258

489

    return true;

1259

489

1260

1261

4

  auto validation_result = header_validator_->validateRequestTrailers(trailers);

1262

4

  std::string failure_details(validation_result.details());

1263

4

  if (validation_result.ok()) {

1264

1

    auto transformation_result = header_validator_->transformRequestTrailers(trailers);

1265

1

    if (transformation_result.ok()) {

1266

      return true;

1267

1268

1

    failure_details = std::string(transformation_result.details());

1269

1

1270

1271

4

  Code response_code = Code::BadRequest;

1272

4

  absl::optional<Grpc::Status::GrpcStatus> grpc_status;

1273

4

  if (Grpc::Common::hasGrpcContentType(*request_headers_)) {

1274

    grpc_status = Grpc::Status::WellKnownGrpcStatus::Internal;

1275

1276

1277

4

  filter_manager_.streamInfo().setResponseFlag(

1278

4

      StreamInfo::CoreResponseFlag::DownstreamProtocolError);

1279

1280

  // H/2 codec was resetting requests that were rejected due to headers with underscores,

1281

  // instead of sending 400. Preserving this behavior for now.

1282

  // TODO(#24466): Make H/2 behavior consistent with H/1 and H/3.

1283

4

  if (failure_details == UhvResponseCodeDetail::get().InvalidUnderscore &&

1284

4

      connection_manager_.codec_->protocol() == Protocol::Http2) {

1285

    filter_manager_.streamInfo().setResponseCodeDetails(failure_details);

1286

    resetStream();

1287

4

  } else {

1288

    // TODO(#24735): Harmonize H/2 and H/3 behavior with H/1

1289

4

    if (connection_manager_.codec_->protocol() < Protocol::Http2) {

1290

2

      sendLocalReply(response_code, "", nullptr, grpc_status, failure_details);

1291

4

    } else {

1292

2

      filter_manager_.streamInfo().setResponseCodeDetails(failure_details);

1293

2

      resetStream();

1294

2

1295

4

    if (!response_encoder_->streamErrorOnInvalidHttpMessage()) {

1296

4

      connection_manager_.handleCodecError(failure_details);

1297

4

1298

4

1299

4

  return false;

1300

4

1301

1302

334657

void ConnectionManagerImpl::ActiveStream::maybeRecordLastByteReceived(bool end_stream) {

1303

  // If recreateStream is called, the HCM rewinds state and may send more encodeData calls.

1304

334657

  if (end_stream && !filter_manager_.hasLastDownstreamByteReceived()) {

1305

84948

    filter_manager_.streamInfo().downstreamTiming().onLastDownstreamRxByteReceived(

1306

84948

        connection_manager_.dispatcher_->timeSource());

1307

84948

    ENVOY_STREAM_LOG(debug, "request end stream timestamp recorded", *this);

1308

84948

1309

334657

1310

1311

// Ordering in this function is complicated, but important.

1312

//

1313

// We want to do minimal work before selecting route and creating a filter

1314

// chain to maximize the number of requests which get custom filter behavior,

1315

// e.g. registering access logging.

1316

//

1317

// This must be balanced by doing sanity checking for invalid requests (one

1318

// can't route select properly without full headers), checking state required to

1319

// serve error responses (connection close, head requests, etc), and

1320

// modifications which may themselves affect route selection.

1321

void ConnectionManagerImpl::ActiveStream::decodeHeaders(RequestHeaderMapSharedPtr&& headers,

1322

91472

                                                        bool end_stream) {

1323

91472

  ENVOY_STREAM_LOG(debug, "request headers complete (end_stream={}):\n{}", *this, end_stream,

1324

91472

                   *headers);

1325

  // We only want to record this when reading the headers the first time, not when recreating

1326

  // a stream.

1327

91472

  if (!filter_manager_.hasLastDownstreamByteReceived()) {

1328

91176

    filter_manager_.streamInfo().downstreamTiming().onLastDownstreamHeaderRxByteReceived(

1329

91176

        connection_manager_.dispatcher_->timeSource());

1330

91176

1331

91472

  ScopeTrackerScopeState scope(this,

1332

91472

                               connection_manager_.read_callbacks_->connection().dispatcher());

1333

91472

  request_headers_ = std::move(headers);

1334

91472

  filter_manager_.requestHeadersInitialized();

1335

91472

  if (request_header_timer_ != nullptr) {

1336

1

    request_header_timer_->disableTimer();

1337

1

    request_header_timer_.reset();

1338

1

1339

1340

  // Both shouldDrainConnectionUponCompletion() and is_head_request_ affect local replies: set them

1341

  // as early as possible.

1342

91472

  const Protocol protocol = connection_manager_.codec_->protocol();

1343

91472

  if (HeaderUtility::shouldCloseConnection(protocol, *request_headers_)) {

1344

    // Only mark the connection to be closed if the request indicates so. The connection might

1345

    // already be marked so before this step, in which case if shouldCloseConnection() returns

1346

    // false, the stream info value shouldn't be overridden.

1347

22

    filter_manager_.streamInfo().setShouldDrainConnectionUponCompletion(true);

1348

22

1349

1350

91472

  filter_manager_.streamInfo().protocol(protocol);

1351

1352

  // We end the decode here to mark that the downstream stream is complete.

1353

91472

  maybeRecordLastByteReceived(end_stream);

1354

1355

91472

  if (!validateHeaders()) {

1356

5

    ENVOY_STREAM_LOG(debug, "request headers validation failed\n{}", *this, *request_headers_);

1357

5

    return;

1358

5

1359

1360

  // We need to snap snapped_route_config_ here as it's used in mutateRequestHeaders later.

1361

91467

  if (connection_manager_.config_->isRoutable()) {

1362

91028

    if (connection_manager_.config_->routeConfigProvider() != nullptr) {

1363

90846

      snapped_route_config_ = connection_manager_.config_->routeConfigProvider()->configCast();

1364

90932

    } else if (connection_manager_.config_->scopedRouteConfigProvider() != nullptr &&

1365

182

               connection_manager_.config_->scopeKeyBuilder().has_value()) {

1366

182

      snapped_scoped_routes_config_ =

1367

182

          connection_manager_.config_->scopedRouteConfigProvider()->config<Router::ScopedConfig>();

1368

182

      snapScopedRouteConfig();

1369

182

1370

91348

  } else {

1371

439

    snapped_route_config_ = connection_manager_.config_->routeConfigProvider()->configCast();

1372

439

1373

1374

  // Drop new requests when overloaded as soon as we have decoded the headers.

1375

91467

  const bool drop_request_due_to_overload =

1376

91467

      (connection_manager_.accept_new_http_stream_ != nullptr &&

1377

91467

       connection_manager_.accept_new_http_stream_->shouldShedLoad()) ||

1378

91467

      connection_manager_.random_generator_.bernoulli(

1379

91459

          connection_manager_.overload_stop_accepting_requests_ref_.value());

1380

1381

91467

  if (drop_request_due_to_overload) {

1382

    // In this one special case, do not create the filter chain. If there is a risk of memory

1383

    // overload it is more important to avoid unnecessary allocation than to create the filters.

1384

49

    filter_manager_.skipFilterChainCreation();

1385

49

    connection_manager_.stats_.named_.downstream_rq_overload_close_.inc();

1386

49

    sendLocalReply(

1387

49

        Http::Code::ServiceUnavailable, "envoy overloaded",

1388

49

        [this](Http::ResponseHeaderMap& headers) {

1389

49

          if (connection_manager_.config_->appendLocalOverload()) {

1390

16

            headers.addReference(Http::Headers::get().EnvoyLocalOverloaded,

1391

16

                                 Http::Headers::get().EnvoyOverloadedValues.True);

1392

16

1393

49

},

1394

49

        absl::nullopt, StreamInfo::ResponseCodeDetails::get().Overload);

1395

49

    return;

1396

49

1397

1398

91418

  if (!connection_manager_.config_->proxy100Continue() && request_headers_->Expect() &&

1399

      // The Expect field-value is case-insensitive.

1400

      // https://tools.ietf.org/html/rfc7231#section-5.1.1

1401

91418

      absl::EqualsIgnoreCase((request_headers_->Expect()->value().getStringView()),

1402

40

                             Headers::get().ExpectValues._100Continue)) {

1403

    // Note in the case Envoy is handling 100-Continue complexity, it skips the filter chain

1404

    // and sends the 100-Continue directly to the encoder.

1405

40

    chargeStats(continueHeader());

1406

40

    response_encoder_->encode1xxHeaders(continueHeader());

1407

    // Remove the Expect header so it won't be handled again upstream.

1408

40

    request_headers_->removeExpect();

1409

40

1410

1411

91418

  connection_manager_.user_agent_.initializeFromHeaders(*request_headers_,

1412

91418

                                                        connection_manager_.stats_.prefixStatName(),

1413

91418

                                                        connection_manager_.stats_.scope_);

1414

1415

91418

  if (!request_headers_->Host()) {

1416

    // Require host header. For HTTP/1.1 Host has already been translated to :authority.

1417

1022

    sendLocalReply(Code::BadRequest, "", nullptr, absl::nullopt,

1418

1022

                   StreamInfo::ResponseCodeDetails::get().MissingHost);

1419

1022

    return;

1420

1022

1421

1422

  // Apply header sanity checks.

1423

90396

  absl::optional<std::reference_wrapper<const absl::string_view>> error =

1424

90396

      HeaderUtility::requestHeadersValid(*request_headers_);

1425

90396

  if (error != absl::nullopt) {

1426

6

    sendLocalReply(Code::BadRequest, "", nullptr, absl::nullopt, error.value().get());

1427

6

    if (!response_encoder_->streamErrorOnInvalidHttpMessage()) {

1428

6

      connection_manager_.handleCodecError(error.value().get());

1429

6

1430

6

    return;

1431

6

1432

1433

  // Check for the existence of the :path header for non-CONNECT requests, or present-but-empty

1434

  // :path header for CONNECT requests. We expect the codec to have broken the path into pieces if

1435

  // applicable. NOTE: Currently the HTTP/1.1 codec only does this when the allow_absolute_url flag

1436

  // is enabled on the HCM.

1437

90390

  if ((!HeaderUtility::isConnect(*request_headers_) || request_headers_->Path()) &&

1438

90390

      request_headers_->getPathValue().empty()) {

1439

2

    sendLocalReply(Code::NotFound, "", nullptr, absl::nullopt,

1440

2

                   StreamInfo::ResponseCodeDetails::get().MissingPath);

1441

2

    return;

1442

2

1443

1444

  // Rewrite the host of CONNECT-UDP requests.

1445

90388

  if (HeaderUtility::isConnectUdpRequest(*request_headers_) &&

1446

90388

      !HeaderUtility::rewriteAuthorityForConnectUdp(*request_headers_)) {

1447

16

    sendLocalReply(Code::NotFound, "The path is incorrect for CONNECT-UDP", nullptr, absl::nullopt,

1448

16

                   StreamInfo::ResponseCodeDetails::get().InvalidPath);

1449

16

    return;

1450

16

1451

1452

  // Currently we only support relative paths at the application layer.

1453

90372

  if (!request_headers_->getPathValue().empty() && request_headers_->getPathValue()[0] != '/') {

1454

2

    connection_manager_.stats_.named_.downstream_rq_non_relative_path_.inc();

1455

2

    sendLocalReply(Code::NotFound, "", nullptr, absl::nullopt,

1456

2

                   StreamInfo::ResponseCodeDetails::get().AbsolutePath);

1457

2

    return;

1458

2

1459

1460

90370

#ifndef ENVOY_ENABLE_UHV

1461

  // In UHV mode path normalization is done in the UHV

1462

  // Path sanitization should happen before any path access other than the above sanity check.

1463

90370

  const auto action =

1464

90370

      ConnectionManagerUtility::maybeNormalizePath(*request_headers_, *connection_manager_.config_);

1465

  // gRPC requests are rejected if Envoy is configured to redirect post-normalization. This is

1466

  // because gRPC clients do not support redirect.

1467

90370

  if (action == ConnectionManagerUtility::NormalizePathAction::Reject ||

1468

90370

      (action == ConnectionManagerUtility::NormalizePathAction::Redirect &&

1469

90328

       Grpc::Common::hasGrpcContentType(*request_headers_))) {

1470

43

    connection_manager_.stats_.named_.downstream_rq_failed_path_normalization_.inc();

1471

43

    sendLocalReply(Code::BadRequest, "", nullptr, absl::nullopt,

1472

43

                   StreamInfo::ResponseCodeDetails::get().PathNormalizationFailed);

1473

43

    return;

1474

90327

  } else if (action == ConnectionManagerUtility::NormalizePathAction::Redirect) {

1475

4

    connection_manager_.stats_.named_.downstream_rq_redirected_with_normalized_path_.inc();

1476

4

    sendLocalReply(

1477

4

        Code::TemporaryRedirect, "",

1478

4

        [new_path = request_headers_->Path()->value().getStringView()](

1479

4

            Http::ResponseHeaderMap& response_headers) -> void {

1480

4

          response_headers.addReferenceKey(Http::Headers::get().Location, new_path);

1481

4

},

1482

4

        absl::nullopt, StreamInfo::ResponseCodeDetails::get().PathNormalizationFailed);

1483

4

    return;

1484

4

1485

1486

90323

  ASSERT(action == ConnectionManagerUtility::NormalizePathAction::Continue);

1487

90323

#endif

1488

90323

  auto optional_port = ConnectionManagerUtility::maybeNormalizeHost(

1489

90323

      *request_headers_, *connection_manager_.config_, localPort());

1490

90323

  if (optional_port.has_value() &&

1491

90323

      requestWasConnect(request_headers_, connection_manager_.codec_->protocol())) {

1492

11

    filter_manager_.streamInfo().filterState()->setData(

1493

11

        Router::OriginalConnectPort::key(),

1494

11

        std::make_unique<Router::OriginalConnectPort>(optional_port.value()),

1495

11

        StreamInfo::FilterState::StateType::ReadOnly, StreamInfo::FilterState::LifeSpan::Request);

1496

11

1497

1498

90323

  if (!state_.is_internally_created_) { // Only sanitize headers on first pass.

1499

    // Modify the downstream remote address depending on configuration and headers.

1500

90027

    const auto mutate_result = ConnectionManagerUtility::mutateRequestHeaders(

1501

90027

        *request_headers_, connection_manager_.read_callbacks_->connection(),

1502

90027

        *connection_manager_.config_, *snapped_route_config_, connection_manager_.local_info_,

1503

90027

        filter_manager_.streamInfo());

1504

1505

    // IP detection failed, reject the request.

1506

90027

    if (mutate_result.reject_request.has_value()) {

1507

1

      const auto& reject_request_params = mutate_result.reject_request.value();

1508

1

      connection_manager_.stats_.named_.downstream_rq_rejected_via_ip_detection_.inc();

1509

1

      sendLocalReply(reject_request_params.response_code, reject_request_params.body, nullptr,

1510

1

                     absl::nullopt,

1511

1

                     StreamInfo::ResponseCodeDetails::get().OriginalIPDetectionFailed);

1512

1

      return;

1513

1

1514

1515

90026

    filter_manager_.setDownstreamRemoteAddress(mutate_result.final_remote_address);

1516

90026

1517

90322

  ASSERT(filter_manager_.streamInfo().downstreamAddressProvider().remoteAddress() != nullptr);

1518

1519

90322

  ASSERT(!cached_route_);

1520

90322

  refreshCachedRoute();

1521

1522

90322

  if (!state_.is_internally_created_) { // Only mutate tracing headers on first pass.

1523

90026

    filter_manager_.streamInfo().setTraceReason(

1524

90026

        ConnectionManagerUtility::mutateTracingRequestHeader(

1525

90026

            *request_headers_, connection_manager_.runtime_, *connection_manager_.config_,

1526

90026

            cached_route_.value().get()));

1527

90026

1528

1529

90322

  filter_manager_.streamInfo().setRequestHeaders(*request_headers_);

1530

90322

  const FilterManager::CreateChainResult create_chain_result =

1531

90322

      filter_manager_.createDownstreamFilterChain();

1532

90322

  if (create_chain_result.upgradeAccepted()) {

1533

601

    connection_manager_.stats_.named_.downstream_cx_upgrades_total_.inc();

1534

601

    connection_manager_.stats_.named_.downstream_cx_upgrades_active_.inc();

1535

601

    state_.successful_upgrade_ = true;

1536

601

1537

1538

90322

  if (connection_manager_.config_->flushAccessLogOnNewRequest()) {

1539

12

    log(AccessLog::AccessLogType::DownstreamStart);

1540

12

1541

1542

  // TODO if there are no filters when starting a filter iteration, the connection manager

1543

  // should return 404. The current returns no response if there is no router filter.

1544

90322

  if (hasCachedRoute()) {

1545

    // Do not allow upgrades if the route does not support it.

1546

87620

    if (create_chain_result.upgradeRejected()) {

1547

      // While downstream servers should not send upgrade payload without the upgrade being

1548

      // accepted, err on the side of caution and refuse to process any further requests on this

1549

      // connection, to avoid a class of HTTP/1.1 smuggling bugs where Upgrade or CONNECT payload

1550

      // contains a smuggled HTTP request.

1551

2

      filter_manager_.streamInfo().setShouldDrainConnectionUponCompletion(true);

1552

2

      connection_manager_.stats_.named_.downstream_rq_ws_on_non_ws_route_.inc();

1553

2

      sendLocalReply(Code::Forbidden, "", nullptr, absl::nullopt,

1554

2

                     StreamInfo::ResponseCodeDetails::get().UpgradeFailed);

1555

2

      return;

1556

2

1557

    // Allow non websocket requests to go through websocket enabled routes.

1558

87620

1559

1560

  // Check if tracing is enabled.

1561

90320

  if (connection_manager_tracing_config_.has_value()) {

1562

298

    traceRequest();

1563

298

1564

1565

90320

  ENVOY_EXECUTION_SCOPE(trackedStream(), active_span_.get());

1566

90320

  if (!connection_manager_.shouldDeferRequestProxyingToNextIoCycle()) {

1567

49892

    filter_manager_.decodeHeaders(*request_headers_, end_stream);

1568

87841

  } else {

1569

40428

    state_.deferred_to_next_io_iteration_ = true;

1570

40428

    state_.deferred_end_stream_ = end_stream;

1571

40428

1572

1573

  // Reset it here for both global and overridden cases.

1574

90320

  resetIdleTimer();

1575

90320

1576

1577

298

void ConnectionManagerImpl::ActiveStream::traceRequest() {

1578

298

  ASSERT(connection_manager_tracing_config_.has_value());

1579

1580

298

  const Tracing::Decision tracing_decision =

1581

298

      Tracing::TracerUtility::shouldTraceRequest(filter_manager_.streamInfo());

1582

1583

298

  if (!trace_refresh_after_route_refresh_) {

1584

1

    ConnectionManagerImpl::chargeTracingStats(tracing_decision.reason,

1585

1

                                              connection_manager_.config_->tracingStats());

1586

1

1587

1588

298

  Tracing::HttpTraceContext trace_context(*request_headers_);

1589

298

  active_span_ = connection_manager_.tracer().startSpan(

1590

298

      *this, trace_context, filter_manager_.streamInfo(), tracing_decision);

1591

1592

298

  if (!active_span_) {

1593

246

    return;

1594

246

1595

52

  if (hasCachedRoute()) {

1596

52

    route_decorator_ = cached_route_.value()->decorator();

1597

52

    route_tracing_ = cached_route_.value()->tracingConfig();

1598

52

1599

52

  if (!operationNameFormatter(*connection_manager_tracing_config_, route_tracing_)) {

1600

    // Only set decorator when there is no operation name formatter configured at either

1601

    // the HCM level or the route level.

1602

49

    setRequestDecorator(*request_headers_);

1603

49

1604

52

1605

1606

242696

void ConnectionManagerImpl::ActiveStream::decodeData(Buffer::Instance& data, bool end_stream) {

1607

242696

  ScopeTrackerScopeState scope(this,

1608

242696

                               connection_manager_.read_callbacks_->connection().dispatcher());

1609

242696

  ENVOY_EXECUTION_SCOPE(trackedStream(), active_span_.get());

1610

242696

  maybeRecordLastByteReceived(end_stream);

1611

242696

  filter_manager_.streamInfo().addBytesReceived(data.length());

1612

242696

  if (!state_.deferred_to_next_io_iteration_) {

1613

242422

    filter_manager_.decodeData(data, end_stream);

1614

242475

  } else {

1615

274

    if (!deferred_data_) {

1616

274

      deferred_data_ = std::make_unique<Buffer::OwnedImpl>();

1617

274

1618

274

    deferred_data_->move(data);

1619

274

    state_.deferred_end_stream_ = end_stream;

1620

274

1621

242696

1622

1623

493

void ConnectionManagerImpl::ActiveStream::decodeTrailers(RequestTrailerMapPtr&& trailers) {

1624

493

  ENVOY_STREAM_LOG(debug, "request trailers complete:\n{}", *this, *trailers);

1625

493

  ScopeTrackerScopeState scope(this,

1626

493

                               connection_manager_.read_callbacks_->connection().dispatcher());

1627

493

  ENVOY_EXECUTION_SCOPE(trackedStream(), active_span_.get());

1628

493

  resetIdleTimer();

1629

1630

493

  ASSERT(!request_trailers_);

1631

493

  if (!validateTrailers(*trailers)) {

1632

4

    ENVOY_STREAM_LOG(debug, "request trailers validation failed:\n{}", *this, *trailers);

1633

4

    return;

1634

4

1635

489

  maybeRecordLastByteReceived(true);

1636

489

  if (!state_.deferred_to_next_io_iteration_) {

1637

345

    request_trailers_ = std::move(trailers);

1638

345

    filter_manager_.decodeTrailers(*request_trailers_);

1639

444

  } else {

1640

    // Save trailers in a different variable since `request_trailers_` is available to the filter

1641

    // manager via `requestTrailers()` callback and makes filter manager see trailers prematurely

1642

    // when deferred request is processed.

1643

144

    deferred_request_trailers_ = std::move(trailers);

1644

144

1645

489

1646

1647

864

void ConnectionManagerImpl::ActiveStream::decodeMetadata(MetadataMapPtr&& metadata_map) {

1648

864

  ENVOY_EXECUTION_SCOPE(trackedStream(), active_span_.get());

1649

864

  resetIdleTimer();

1650

864

  if (!state_.deferred_to_next_io_iteration_) {

1651

    // After going through filters, the ownership of metadata_map will be passed to terminal filter.

1652

    // The terminal filter may encode metadata_map to the next hop immediately or store metadata_map

1653

    // and encode later when connection pool is ready.

1654

806

    filter_manager_.decodeMetadata(*metadata_map);

1655

826

  } else {

1656

58

    deferred_metadata_.push(std::move(metadata_map));

1657

58

1658

864

1659

1660

192605

void ConnectionManagerImpl::ActiveStream::disarmRequestTimeout() {

1661

192605

  if (request_timer_) {

1662

172

    request_timer_->disableTimer();

1663

172

1664

192605

1665

1666

204

void ConnectionManagerImpl::startDrainSequence() {

1667

204

  ASSERT(drain_state_ == DrainState::NotDraining);

1668

204

  drain_state_ = DrainState::Draining;

1669

204

  codec_->shutdownNotice();

1670

204

  drain_timer_ = dispatcher_->createTimer([this]() -> void { onDrainTimeout(); });

1671

204

  drain_timer_->enableTimer(config_->drainTimeout());

1672

204

1673

1674

406

void ConnectionManagerImpl::ActiveStream::snapScopedRouteConfig() {

1675

  // NOTE: if a RDS subscription hasn't got a RouteConfiguration back, a Router::NullConfigImpl is

1676

  // returned, in that case we let it pass.

1677

406

  auto scope_key =

1678

406

      connection_manager_.config_->scopeKeyBuilder()->computeScopeKey(*request_headers_);

1679

406

  snapped_route_config_ = snapped_scoped_routes_config_->getRouteConfig(scope_key);

1680

406

  if (snapped_route_config_ == nullptr) {

1681

158

    ENVOY_STREAM_LOG(trace, "can't find SRDS scope.", *this);

1682

    // TODO(stevenzzzz): Consider to pass an error message to router filter, so that it can

1683

    // send back 404 with some more details.

1684

158

    snapped_route_config_ = std::make_shared<Router::NullConfigImpl>();

1685

158

1686

406

1687

1688

90378

void ConnectionManagerImpl::ActiveStream::refreshCachedRoute() { refreshCachedRoute(nullptr); }

1689

1690

93452

void ConnectionManagerImpl::ActiveStream::refreshDurationTimeout() {

1691

93452

  if (!hasCachedRoute() || !request_headers_) {

1692

5205

    return;

1693

5205

1694

88247

  const Router::RouteEntry* route = cached_route_.value()->routeEntry();

1695

88247

  if (route == nullptr) {

1696

276

    return;

1697

276

1698

87971

  auto grpc_timeout = Grpc::Common::getGrpcTimeout(*request_headers_);

1699

87971

  std::chrono::milliseconds timeout;

1700

87971

  bool disable_timer = false;

1701

1702

87971

  if (!grpc_timeout || !route->grpcTimeoutHeaderMax()) {

1703

    // Either there is no grpc-timeout header or special timeouts for it are not

1704

    // configured. Use stream duration.

1705

87957

    if (route->maxStreamDuration()) {

1706

2

      timeout = route->maxStreamDuration().value();

1707

2

      if (timeout == std::chrono::milliseconds(0)) {

1708

        // Explicitly configured 0 means no timeout.

1709

        disable_timer = true;

1710

1711

87955

    } else {

1712

      // Fall back to HCM config. If no HCM duration limit exists, disable

1713

      // timers set by any prior route configuration.

1714

87955

      const auto max_stream_duration = connection_manager_.config_->maxStreamDuration();

1715

87955

      if (max_stream_duration.has_value() && max_stream_duration.value().count()) {

1716

36

        timeout = max_stream_duration.value();

1717

87919

      } else {

1718

87919

        disable_timer = true;

1719

87919

1720

87955

1721

87957

  } else {

1722

    // Start with the timeout equal to the gRPC timeout header.

1723

14

    timeout = grpc_timeout.value();

1724

    // If there's a valid cap, apply it.

1725

14

    if (timeout > route->grpcTimeoutHeaderMax().value() &&

1726

14

        route->grpcTimeoutHeaderMax().value() != std::chrono::milliseconds(0)) {

1727

5

      timeout = route->grpcTimeoutHeaderMax().value();

1728

5

1729

1730

    // Apply the configured offset.

1731

14

    if (timeout != std::chrono::milliseconds(0) && route->grpcTimeoutHeaderOffset()) {

1732

2

      const auto offset = route->grpcTimeoutHeaderOffset().value();

1733

2

      if (offset < timeout) {

1734

1

        timeout -= offset;

1735

1

      } else {

1736

1

        timeout = std::chrono::milliseconds(0);

1737

1

1738

2

1739

14

1740

1741

  // Disable any existing timer if configured to do so.

1742

87971

  if (disable_timer) {

1743

87919

    if (max_stream_duration_timer_) {

1744

2

      max_stream_duration_timer_->disableTimer();

1745

2

      if (route->usingNewTimeouts() && Grpc::Common::isGrpcRequestHeaders(*request_headers_)) {

1746

        request_headers_->removeGrpcTimeout();

1747

1748

2

1749

87919

    return;

1750

87919

1751

1752

  // Set the header timeout before doing used-time adjustments.

1753

  // This may result in the upstream not getting the latest results, but also

1754

  // avoids every request getting a custom timeout based on envoy think time.

1755

52

  if (route->usingNewTimeouts() && Grpc::Common::isGrpcRequestHeaders(*request_headers_)) {

1756

6

    Grpc::Common::toGrpcTimeout(std::chrono::milliseconds(timeout), *request_headers_);

1757

6

1758

1759

  // See how long this stream has been alive, and adjust the timeout

1760

  // accordingly.

1761

52

  std::chrono::milliseconds time_used = std::chrono::duration_cast<std::chrono::milliseconds>(

1762

52

      connection_manager_.timeSource().monotonicTime() -

1763

52

      filter_manager_.streamInfo().startTimeMonotonic());

1764

52

  if (timeout > time_used) {

1765

49

    timeout -= time_used;

1766

49

  } else {

1767

3

    timeout = std::chrono::milliseconds(0);

1768

3

1769

1770

  // Finally create (if necessary) and enable the timer.

1771

52

  if (!max_stream_duration_timer_) {

1772

7

    max_stream_duration_timer_ = connection_manager_.dispatcher_->createTimer(

1773

7

        [this]() -> void { onStreamMaxDurationReached(); });

1774

7

1775

52

  max_stream_duration_timer_->enableTimer(timeout);

1776

52

1777

1778

93436

void ConnectionManagerImpl::ActiveStream::refreshCachedRoute(const Router::RouteCallback& cb) {

1779

  // If the cached route is blocked then any attempt to clear it or refresh it

1780

  // will be ignored.

1781

93436

  if (routeCacheBlocked()) {

1782

    return;

1783

1784

1785

93436

  Router::VirtualHostRoute route_result;

1786

93436

  if (request_headers_ != nullptr) {

1787

92087

    if (connection_manager_.config_->isRoutable() &&

1788

92087

        connection_manager_.config_->scopedRouteConfigProvider() != nullptr &&

1789

92087

        connection_manager_.config_->scopeKeyBuilder().has_value()) {

1790

      // NOTE: re-select scope as well in case the scope key header has been changed by a filter.

1791

224

      snapScopedRouteConfig();

1792

224

1793

92087

    if (snapped_route_config_ != nullptr) {

1794

92082

      route_result = snapped_route_config_->route(cb, *request_headers_,

1795

92082

                                                  filter_manager_.streamInfo(), stream_id_);

1796

92082

1797

92087

1798

1799

93436

  setVirtualHostRoute(std::move(route_result));

1800

93436

1801

1802

93452

void ConnectionManagerImpl::ActiveStream::refreshTracing() {

1803

93452

  if (!trace_refresh_after_route_refresh_) {

1804

2

    return;

1805

2

1806

1807

93450

  if (!connection_manager_tracing_config_.has_value() || active_span_ == nullptr ||

1808

93450

      request_headers_ == nullptr) {

1809

93448

    return;

1810

93448

1811

1812

2

  ASSERT(cached_route_.has_value());

1813

1814

  // NOTE: if the trace reason have been encoded into the request id then the trace reason may

1815

  // not be updated. That means we may cannot to force a traced request to be untraced by the

1816

  // refreshing.

1817

2

  const auto trace_reason = ConnectionManagerUtility::mutateTracingRequestHeader(

1818

2

      *request_headers_, connection_manager_.runtime_, *connection_manager_.config_,

1819

2

      cached_route_.value().get());

1820

2

  filter_manager_.streamInfo().setTraceReason(trace_reason);

1821

2

  const Tracing::Decision tracing_decision =

1822

2

      Tracing::TracerUtility::shouldTraceRequest(filter_manager_.streamInfo());

1823

2

  if (active_span_->useLocalDecision()) {

1824

1

    active_span_->setSampled(tracing_decision.traced);

1825

1

1826

1827

2

  if (hasCachedRoute()) {

1828

2

    route_decorator_ = cached_route_.value()->decorator();

1829

2

    route_tracing_ = cached_route_.value()->tracingConfig();

1830

2

1831

2

  if (!operationNameFormatter(*connection_manager_tracing_config_, route_tracing_)) {

1832

    // Only set decorator when there is no operation name formatter configured at either

1833

    // the HCM level or the route level.

1834

2

    setRequestDecorator(*request_headers_);

1835

2

1836

2

1837

1838

// TODO(chaoqin-li1123): Make on demand vhds and on demand srds works at the same time.

1839

void ConnectionManagerImpl::ActiveStream::requestRouteConfigUpdate(

1840

76

    Http::RouteConfigUpdatedCallbackSharedPtr route_config_updated_cb) {

1841

76

  ENVOY_BUG(route_config_update_requester_.has_value(),

1842

76

            "RouteConfigUpdate requested but RDS not compiled into the binary. Try linking "

1843

76

            "//source/common/http:rds_lib");

1844

76

  if (route_config_update_requester_.has_value()) {

1845

76

    (*route_config_update_requester_)

1846

76

        ->requestRouteConfigUpdate(*this, route_config_updated_cb, routeConfig(),

1847

76

                                   *connection_manager_.dispatcher_, *request_headers_);

1848

76

1849

76

1850

1851

76

absl::optional<Router::ConfigConstSharedPtr> ConnectionManagerImpl::ActiveStream::routeConfig() {

1852

76

  if (connection_manager_.config_->routeConfigProvider() != nullptr) {

1853

24

    return {connection_manager_.config_->routeConfigProvider()->configCast()};

1854

24

1855

52

  return {};

1856

76

1857

1858

10124

void ConnectionManagerImpl::ActiveStream::onLocalReply(Code code) {

1859

  // The BadRequest error code indicates there has been a messaging error.

1860

10124

  if (code == Http::Code::BadRequest && connection_manager_.codec_->protocol() < Protocol::Http2 &&

1861

10124

      !response_encoder_->streamErrorOnInvalidHttpMessage()) {

1862

1374

    filter_manager_.streamInfo().setShouldDrainConnectionUponCompletion(true);

1863

1374

1864

10124

1865

1866

129

void ConnectionManagerImpl::ActiveStream::encode1xxHeaders(ResponseHeaderMap& response_headers) {

1867

129

  ENVOY_EXECUTION_SCOPE(trackedStream(), active_span_.get());

1868

  // Strip the T-E headers etc. Defer other header additions as well as drain-close logic to the

1869

  // continuation headers.

1870

129

  ConnectionManagerUtility::mutateResponseHeaders(

1871

129

      response_headers, request_headers_.get(), *connection_manager_.config_, EMPTY_STRING,

1872

129

      filter_manager_.streamInfo(), connection_manager_.proxy_name_,

1873

129

      connection_manager_.clear_hop_by_hop_response_headers_);

1874

1875

  // Count both the 1xx and follow-up response code in stats.

1876

129

  chargeStats(response_headers);

1877

1878

129

  ENVOY_STREAM_LOG(debug, "encoding 1xx continue headers via codec:\n{}", *this, response_headers);

1879

1880

  // Now actually encode via the codec.

1881

129

  response_encoder_->encode1xxHeaders(response_headers);

1882

129

1883

1884

void ConnectionManagerImpl::ActiveStream::encodeHeaders(ResponseHeaderMap& headers,

1885

47257

                                                        bool end_stream) {

1886

47257

  ENVOY_EXECUTION_SCOPE(trackedStream(), active_span_.get());

1887

  // Base headers.

1888

1889

  // We want to preserve the original date header, but we add a date header if it is absent

1890

47257

  if (!headers.Date()) {

1891

46900

    connection_manager_.config_->dateProvider().setDateHeader(headers);

1892

46900

1893

1894

  // Following setReference() is safe because serverName() is constant for the life of the

1895

  // listener.

1896

47257

  const auto transformation = connection_manager_.config_->serverHeaderTransformation();

1897

47257

  if (transformation == ConnectionManagerConfig::HttpConnectionManagerProto::OVERWRITE ||

1898

47257

      (transformation == ConnectionManagerConfig::HttpConnectionManagerProto::APPEND_IF_ABSENT &&

1899

47254

       headers.Server() == nullptr)) {

1900

47254

    headers.setReferenceServer(connection_manager_.config_->serverName());

1901

47254

1902

47257

  ConnectionManagerUtility::mutateResponseHeaders(

1903

47257

      headers, request_headers_.get(), *connection_manager_.config_,

1904

47257

      connection_manager_.config_->via(), filter_manager_.streamInfo(),

1905

47257

      connection_manager_.proxy_name_, connection_manager_.clear_hop_by_hop_response_headers_);

1906

1907

47257

  bool drain_connection_due_to_overload = false;

1908

47257

  if (connection_manager_.drain_state_ == DrainState::NotDraining &&

1909

47257

      connection_manager_.random_generator_.bernoulli(

1910

47191

          connection_manager_.overload_disable_keepalive_ref_.value())) {

1911

6

    ENVOY_STREAM_LOG(debug, "disabling keepalive due to envoy overload", *this);

1912

6

    drain_connection_due_to_overload = true;

1913

6

    connection_manager_.stats_.named_.downstream_cx_overload_disable_keepalive_.inc();

1914

6

1915

1916

  // If we're an inbound listener, then we should drain even if the drain direction is inbound only.

1917

  // If not, then we only drain if the drain direction is all.

1918

47257

  Network::DrainDirection drain_scope =

1919

47257

      connection_manager_.direction_ == envoy::config::core::v3::TrafficDirection::INBOUND

1920

47257

          ? Network::DrainDirection::InboundOnly

1921

47257

          : Network::DrainDirection::All;

1922

1923

  // See if we want to drain/close the connection. Send the go away frame prior to encoding the

1924

  // header block. Only drain if the drain direction is not inbound only or the connection is

1925

  // inbound.

1926

47257

  if (connection_manager_.drain_state_ == DrainState::NotDraining &&

1927

47257

      (connection_manager_.drain_close_.drainClose(drain_scope) ||

1928

47191

       drain_connection_due_to_overload)) {

1929

1930

    // This doesn't really do anything for HTTP/1.1 other then give the connection another boost

1931

    // of time to race with incoming requests. For HTTP/2 connections, send a GOAWAY frame to

1932

    // prevent any new streams.

1933

69

    connection_manager_.startDrainSequence();

1934

69

    connection_manager_.stats_.named_.downstream_cx_drain_close_.inc();

1935

69

    ENVOY_STREAM_LOG(debug, "drain closing connection", *this);

1936

69

1937

1938

47257

  if (connection_manager_.codec_->protocol() == Protocol::Http10) {

1939

    // As HTTP/1.0 and below can not do chunked encoding, if there is no content

1940

    // length the response will be framed by connection close.

1941

18

    if (!headers.ContentLength()) {

1942

10

      filter_manager_.streamInfo().setShouldDrainConnectionUponCompletion(true);

1943

10

1944

    // If the request came with a keep-alive and no other factor resulted in a

1945

    // connection close header, send an explicit keep-alive header.

1946

18

    if (!filter_manager_.streamInfo().shouldDrainConnectionUponCompletion()) {

1947

8

      headers.setConnection(Headers::get().ConnectionValues.KeepAlive);

1948

8

1949

18

1950

1951

47257

  if (connection_manager_.drain_state_ == DrainState::NotDraining &&

1952

47257

      filter_manager_.streamInfo().shouldDrainConnectionUponCompletion()) {

1953

1414

    ENVOY_STREAM_LOG(debug, "closing connection due to connection close header", *this);

1954

    // For HTTP/2 and HTTP/3, send GOAWAY and allow current stream to complete.

1955

    // For HTTP/1.1, go directly to Closing (Connection: close header will be added below).

1956

1414

    if (connection_manager_.codec_->protocol() >= Protocol::Http2) {

1957

8

      connection_manager_.startDrainSequence();

1958

1406

    } else {

1959

1406

      connection_manager_.drain_state_ = DrainState::Closing;

1960

1406

1961

1414

1962

1963

  // If we are destroying a stream before remote is complete and the connection does not support

1964

  // multiplexing, we should disconnect since we don't want to wait around for the request to

1965

  // finish.

1966

47257

  if (!filter_manager_.hasLastDownstreamByteReceived()) {

1967

2951

    if (connection_manager_.codec_->protocol() < Protocol::Http2) {

1968

1921

      connection_manager_.drain_state_ = DrainState::Closing;

1969

1921

1970

1971

2951

    connection_manager_.stats_.named_.downstream_rq_response_before_rq_complete_.inc();

1972

2951

1973

1974

47257

  if (Utility::isUpgrade(headers) ||

1975

47257

      HeaderUtility::isConnectResponse(request_headers_.get(), *responseHeaders())) {

1976

341

    state_.is_tunneling_ = true;

1977

341

1978

1979

  // Block route cache if the response headers is received and processed. Because after this

1980

  // point, the cached route should never be updated or refreshed.

1981

47257

  blockRouteCache();

1982

1983

47257

  if (connection_manager_.drain_state_ != DrainState::NotDraining &&

1984

47257

      connection_manager_.codec_->protocol() < Protocol::Http2) {

1985

    // If the connection manager is draining send "Connection: Close" on HTTP/1.1 connections.

1986

    // Do not do this for H2 (which drains via GOAWAY) or Upgrade or CONNECT (as the

1987

    // payload is no longer HTTP/1.1)

1988

2051

    if (!state_.is_tunneling_) {

1989

1982

      headers.setReferenceConnection(Headers::get().ConnectionValues.Close);

1990

1982

1991

2051

1992

1993

47257

  if (active_span_ != nullptr) {

1994

52

    ASSERT(connection_manager_tracing_config_.has_value());

1995

52

    if (!operationNameFormatter(*connection_manager_tracing_config_, route_tracing_)) {

1996

      // Only apply decorator if there is no operation name formatter configured at either

1997

      // the HCM level or the route level.

1998

49

      setResponseDecorator(headers);

1999

49

2000

52

2001

2002

47257

  chargeStats(headers);

2003

2004

47257

  if (state_.is_tunneling_ &&

2005

47257

      connection_manager_.config_->flushAccessLogOnTunnelSuccessfullyEstablished()) {

2006

5

    log(AccessLog::AccessLogType::DownstreamTunnelSuccessfullyEstablished);

2007

5

2008

47257

  ENVOY_STREAM_LOG(debug, "encoding headers via codec (end_stream={}):\n{}", *this, end_stream,

2009

47257

                   headers);

2010

2011

47257

  filter_manager_.streamInfo().downstreamTiming().onFirstDownstreamTxByteSent(

2012

47257

      connection_manager_.time_source_);

2013

2014

47257

  if (header_validator_) {

2015

9

    auto result = header_validator_->transformResponseHeaders(headers);

2016

9

    if (!result.status.ok()) {

2017

      // It is possible that the header map is invalid if an encoder filter makes invalid

2018

      // modifications

2019

      // TODO(yanavlasov): add handling for this case.

2020

9

    } else if (result.new_headers) {

2021

      response_encoder_->encodeHeaders(*result.new_headers, end_stream);

2022

      return;

2023

2024

9

2025

2026

  // Now actually encode via the codec.

2027

47257

  response_encoder_->encodeHeaders(headers, end_stream);

2028

47257

2029

2030

70894

void ConnectionManagerImpl::ActiveStream::encodeData(Buffer::Instance& data, bool end_stream) {

2031

70894

  ENVOY_EXECUTION_SCOPE(trackedStream(), active_span_.get());

2032

70894

  ENVOY_STREAM_LOG(trace, "encoding data via codec (size={} end_stream={})", *this, data.length(),

2033

70894

                   end_stream);

2034

2035

70894

  filter_manager_.streamInfo().addBytesSent(data.length());

2036

70894

  response_encoder_->encodeData(data, end_stream);

2037

70894

2038

2039

448

void ConnectionManagerImpl::ActiveStream::encodeTrailers(ResponseTrailerMap& trailers) {

2040

448

  ENVOY_EXECUTION_SCOPE(trackedStream(), active_span_.get());

2041

448

  ENVOY_STREAM_LOG(debug, "encoding trailers via codec:\n{}", *this, trailers);

2042

2043

448

  response_encoder_->encodeTrailers(trailers);

2044

448

2045

2046

1590

void ConnectionManagerImpl::ActiveStream::encodeMetadata(MetadataMapPtr&& metadata) {

2047

1590

  ENVOY_EXECUTION_SCOPE(trackedStream(), active_span_.get());

2048

1590

  MetadataMapVector metadata_map_vector;

2049

1590

  metadata_map_vector.emplace_back(std::move(metadata));

2050

1590

  ENVOY_STREAM_LOG(debug, "encoding metadata via codec:\n{}", *this, metadata_map_vector);

2051

1590

  response_encoder_->encodeMetadata(metadata_map_vector);

2052

1590

2053

2054

220406

void ConnectionManagerImpl::ActiveStream::onDecoderFilterBelowWriteBufferLowWatermark() {

2055

220406

  ENVOY_STREAM_LOG(debug, "Read-enabling downstream stream due to filter callbacks.", *this);

2056

  // If the state is destroyed, the codec's stream is already torn down. On

2057

  // teardown the codec will unwind any remaining read disable calls.

2058

220406

  if (!filter_manager_.destroyed()) {

2059

220404

    response_encoder_->getStream().readDisable(false);

2060

220404

2061

220406

  connection_manager_.stats_.named_.downstream_flow_control_resumed_reading_total_.inc();

2062

220406

2063

2064

220408

void ConnectionManagerImpl::ActiveStream::onDecoderFilterAboveWriteBufferHighWatermark() {

2065

220408

  ENVOY_STREAM_LOG(debug, "Read-disabling downstream stream due to filter callbacks.", *this);

2066

220408

  response_encoder_->getStream().readDisable(true);

2067

220408

  connection_manager_.stats_.named_.downstream_flow_control_paused_reading_total_.inc();

2068

220408

2069

2070

void ConnectionManagerImpl::ActiveStream::onResetStream(StreamResetReason reset_reason,

2071

49630

                                                        absl::string_view) {

2072

  // NOTE: This function gets called in all of the following cases:

2073

  //       1) We TX an app level reset

2074

  //       2) The codec TX a codec level reset

2075

  //       3) The codec RX a reset

2076

  //       4) The overload manager reset the stream

2077

  //       If we need to differentiate we need to do it inside the codec. Can start with this.

2078

49630

  const absl::string_view encoder_details = response_encoder_->getStream().responseDetails();

2079

49630

  ENVOY_STREAM_LOG(debug, "stream reset: reset reason: {}, response details: {}", *this,

2080

49630

                   Http::Utility::resetReasonToString(reset_reason),

2081

49630

                   encoder_details.empty() ? absl::string_view{"-"} : encoder_details);

2082

49630

  connection_manager_.stats_.named_.downstream_rq_rx_reset_.inc();

2083

49630

  state_.on_reset_stream_called_ = true;

2084

2085

  // If the codec sets its responseDetails() for a reason other than peer reset, set a

2086

  // DownstreamProtocolError. Either way, propagate details.

2087

49630

  if (reset_reason == StreamResetReason::ProtocolError ||

2088

49630

      (!encoder_details.empty() && reset_reason == StreamResetReason::LocalReset)) {

2089

2282

    filter_manager_.streamInfo().setResponseFlag(

2090

2282

        StreamInfo::CoreResponseFlag::DownstreamProtocolError);

2091

2282

2092

49630

  if (!encoder_details.empty()) {

2093

3629

    if (reset_reason == StreamResetReason::ConnectError ||

2094

3629

        reset_reason == StreamResetReason::RemoteReset ||

2095

3629

        reset_reason == StreamResetReason::RemoteRefusedStreamReset) {

2096

1055

      filter_manager_.streamInfo().setResponseFlag(

2097

1055

          StreamInfo::CoreResponseFlag::DownstreamRemoteReset);

2098

1055

2099

3629

    filter_manager_.streamInfo().setResponseCodeDetails(encoder_details);

2100

3629

2101

2102

  // Check if we're in the overload manager reset case.

2103

  // encoder_details should be empty in this case as we don't have a codec error.

2104

49630

  if (encoder_details.empty() && reset_reason == StreamResetReason::OverloadManager) {

2105

12

    filter_manager_.streamInfo().setResponseFlag(StreamInfo::CoreResponseFlag::OverloadManager);

2106

12

    filter_manager_.streamInfo().setResponseCodeDetails(

2107

12

        StreamInfo::ResponseCodeDetails::get().Overload);

2108

12

2109

49630

  filter_manager_.onDownstreamReset();

2110

49630

  connection_manager_.doDeferredStreamDestroy(*this);

2111

49630

2112

2113

19341

void ConnectionManagerImpl::ActiveStream::onAboveWriteBufferHighWatermark() {

2114

19341

  ENVOY_STREAM_LOG(debug, "Disabling upstream stream due to downstream stream watermark.", *this);

2115

19341

  filter_manager_.callHighWatermarkCallbacks();

2116

19341

2117

2118

19189

void ConnectionManagerImpl::ActiveStream::onBelowWriteBufferLowWatermark() {

2119

19189

  ENVOY_STREAM_LOG(debug, "Enabling upstream stream due to downstream stream watermark.", *this);

2120

19189

  filter_manager_.callLowWatermarkCallbacks();

2121

19189

2122

2123

49927

bool ConnectionManagerImpl::ActiveStream::shouldSkipDeferredCloseDelay() const {

2124

  // If HTTP/1.0 has no content length, it is framed by close and won't consider

2125

  // the request complete until the FIN is read. Don't delay close in this case.

2126

49927

  const bool http_10_sans_cl = (connection_manager_.codec_->protocol() == Protocol::Http10) &&

2127

49927

                               (!response_headers_ || !response_headers_->ContentLength());

2128

  // We also don't delay-close in the case of HTTP/1.1 where the request is

2129

  // fully read, as there's no race condition to avoid.

2130

49927

  const bool connection_close = filter_manager_.streamInfo().shouldDrainConnectionUponCompletion();

2131

49927

  const bool request_complete = filter_manager_.hasLastDownstreamByteReceived();

2132

2133

  // Don't do delay close for HTTP/1.0 or if the request is complete.

2134

49927

  return connection_close && (request_complete || http_10_sans_cl);

2135

49927

2136

2137

46122

void ConnectionManagerImpl::ActiveStream::onCodecEncodeComplete() {

2138

46122

  ASSERT(!state_.codec_encode_complete_);

2139

46122

  ENVOY_STREAM_LOG(debug, "Codec completed encoding stream.", *this);

2140

46122

  state_.codec_encode_complete_ = true;

2141

2142

  // Update timing

2143

46122

  filter_manager_.streamInfo().downstreamTiming().onLastDownstreamTxByteSent(

2144

46122

      connection_manager_.time_source_);

2145

46122

  request_response_timespan_->complete();

2146

2147

  // Only reap stream once.

2148

46122

  if (state_.is_zombie_stream_) {

2149

2801

    const bool skip_delay = shouldSkipDeferredCloseDelay();

2150

2801

    connection_manager_.doDeferredStreamDestroy(*this);

2151

    // After destroying a zombie stream, check if the connection should be

2152

    // closed. doEndStream() call that created the zombie may have set

2153

    // drain_state_ to Closing, but checkForDeferredClose() couldn't close the

2154

    // connection at that time because streams_ wasn't empty yet.

2155

2801

    if (connection_manager_.close_connection_on_zombie_stream_complete_) {

2156

2800

      connection_manager_.checkForDeferredClose(skip_delay);

2157

2800

2158

2801

2159

46122

2160

2161

446

void ConnectionManagerImpl::ActiveStream::onCodecLowLevelReset() {

2162

446

  ASSERT(!state_.codec_encode_complete_);

2163

446

  state_.on_reset_stream_called_ = true;

2164

446

  ENVOY_STREAM_LOG(debug, "Codec low level reset", *this);

2165

2166

  // TODO(kbaichoo): Update streamInfo to account for the reset.

2167

2168

  // Only reap stream once.

2169

446

  if (state_.is_zombie_stream_) {

2170

180

    const bool skip_delay = shouldSkipDeferredCloseDelay();

2171

180

    connection_manager_.doDeferredStreamDestroy(*this);

2172

    // After destroying a zombie stream, check if the connection should be

2173

    // closed. doEndStream() call that created the zombie may have set

2174

    // drain_state_ to Closing, but checkForDeferredClose() couldn't close the

2175

    // connection at that time because streams_ wasn't empty yet.

2176

180

    if (connection_manager_.close_connection_on_zombie_stream_complete_) {

2177

180

      connection_manager_.checkForDeferredClose(skip_delay);

2178

180

2179

180

2180

446

2181

2182

107

Tracing::OperationName ConnectionManagerImpl::ActiveStream::operationName() const {

2183

107

  ASSERT(connection_manager_tracing_config_.has_value());

2184

107

  return connection_manager_tracing_config_->operation_name_;

2185

107

2186

2187

void ConnectionManagerImpl::ActiveStream::modifySpan(Tracing::Span& span,

2188

69

                                                     bool upstream_span) const {

2189

69

  ASSERT(connection_manager_tracing_config_.has_value());

2190

2191

69

  const Tracing::HttpTraceContext trace_context(*request_headers_);

2192

69

  const Formatter::Context formatter_context{

2193

69

      request_headers_.get(), response_headers_.get(), response_trailers_.get(), {}, {},

2194

69

      active_span_.get()};

2195

69

  const Tracing::CustomTagContext ctx{trace_context, filter_manager_.streamInfo(),

2196

69

                                      formatter_context};

2197

2198

  // Cache the optional custom tags from the route first.

2199

69

  OptRef<const Tracing::CustomTagMap> route_custom_tags;

2200

2201

69

  if (route_tracing_ != nullptr) {

2202

7

    route_custom_tags.emplace(route_tracing_->getCustomTags());

2203

12

    for (const auto& tag : *route_custom_tags) {

2204

8

      tag.second->applySpan(span, ctx);

2205

8

2206

7

2207

2208

77

  for (const auto& tag : connection_manager_tracing_config_->custom_tags_) {

2209

24

    if (!route_custom_tags.has_value() || !route_custom_tags->contains(tag.first)) {

2210

      // If the tag is defined in both the connection manager and the route,

2211

      // use the route's tag.

2212

20

      tag.second->applySpan(span, ctx);

2213

20

2214

24

2215

2216

  // For same stream, there is only one downstream span. It's the active span.

2217

  // So we can determine whether the span is downstream span by comparing the

2218

  // span pointer.

2219

69

  const Formatter::Formatter* operation =

2220

69

      upstream_span

2221

69

          ? upstreamOperationNameFormatter(*connection_manager_tracing_config_, route_tracing_)

2222

69

          : operationNameFormatter(*connection_manager_tracing_config_, route_tracing_);

2223

69

  if (operation != nullptr) {

2224

6

    const auto op = operation->format(formatter_context, filter_manager_.streamInfo());

2225

6

    if (!op.empty()) {

2226

6

      span.setOperation(op);

2227

6

2228

6

2229

69

2230

2231

66

bool ConnectionManagerImpl::ActiveStream::verbose() const {

2232

66

  ASSERT(connection_manager_tracing_config_.has_value());

2233

66

  return connection_manager_tracing_config_->verbose_;

2234

66

2235

2236

52

uint32_t ConnectionManagerImpl::ActiveStream::maxPathTagLength() const {

2237

52

  ASSERT(connection_manager_tracing_config_.has_value());

2238

52

  return connection_manager_tracing_config_->max_path_tag_length_;

2239

52

2240

2241

68

bool ConnectionManagerImpl::ActiveStream::spawnUpstreamSpan() const {

2242

68

  ASSERT(connection_manager_tracing_config_.has_value());

2243

68

  return connection_manager_tracing_config_->spawn_upstream_span_;

2244

68

2245

2246

35

bool ConnectionManagerImpl::ActiveStream::noContextPropagation() const {

2247

35

  ASSERT(connection_manager_tracing_config_.has_value());

2248

35

  return connection_manager_tracing_config_->no_context_propagation_;

2249

35

2250

2251

652

const Router::RouteEntry::UpgradeMap* ConnectionManagerImpl::ActiveStream::upgradeMap() {

2252

  // We must check if the 'cached_route_' optional is populated since this function can be called

2253

  // early via sendLocalReply(), before the cached route is populated.

2254

652

  if (hasCachedRoute() && cached_route_.value()->routeEntry()) {

2255

597

    return &cached_route_.value()->routeEntry()->upgradeMap();

2256

597

2257

2258

55

  return nullptr;

2259

652

2260

2261

131201

Tracing::Span& ConnectionManagerImpl::ActiveStream::activeSpan() {

2262

131201

  if (active_span_) {

2263

128

    return *active_span_;

2264

131172

  } else {

2265

131073

    return Tracing::NullSpan::instance();

2266

131073

2267

131201

2268

2269

43676

OptRef<const Tracing::Config> ConnectionManagerImpl::ActiveStream::tracingConfig() const {

2270

43676

  if (connection_manager_tracing_config_.has_value()) {

2271

52

    return makeOptRef<const Tracing::Config>(*this);

2272

52

2273

43624

  return {};

2274

43676

2275

2276

164795

const ScopeTrackedObject& ConnectionManagerImpl::ActiveStream::scope() { return *this; }

2277

2278

671

Upstream::ClusterInfoConstSharedPtr ConnectionManagerImpl::ActiveStream::clusterInfo() {

2279

  // NOTE: Refreshing route caches clusterInfo as well.

2280

671

  if (!cached_route_.has_value()) {

2281

16

    refreshCachedRoute();

2282

16

2283

2284

671

  return cached_cluster_info_.value();

2285

671

2286

2287

Router::RouteConstSharedPtr

2288

151243

ConnectionManagerImpl::ActiveStream::route(const Router::RouteCallback& cb) {

2289

151243

  if (cached_route_.has_value()) {

2290

148185

    return cached_route_.value();

2291

148185

2292

3058

  refreshCachedRoute(cb);

2293

3058

  return cached_route_.value();

2294

151243

2295

2296

17

void ConnectionManagerImpl::ActiveStream::setRoute(Router::RouteConstSharedPtr route) {

2297

17

  Router::VirtualHostRoute vhost_route;

2298

17

  if (route != nullptr) {

2299

15

    vhost_route.vhost = route->virtualHost();

2300

15

    vhost_route.route = std::move(route);

2301

15

2302

17

  setVirtualHostRoute(std::move(vhost_route));

2303

17

2304

2305

/**

2306

 * Sets the cached route to the RouteConstSharedPtr argument passed in. Handles setting the

2307

 * cached_route_/cached_cluster_info_ ActiveStream attributes, the FilterManager streamInfo, tracing

2308

 * tags, and timeouts.

2309

2310

 * Declared as a StreamFilterCallbacks member function for filters to call directly, but also

2311

 * functions as a helper to refreshCachedRoute(const Router::RouteCallback& cb).

2312

*/

2313

void ConnectionManagerImpl::ActiveStream::setVirtualHostRoute(

2314

93453

    Router::VirtualHostRoute vhost_route) {

2315

  // If the cached route is blocked then any attempt to clear it or refresh it

2316

  // will be ignored.

2317

93453

  if (routeCacheBlocked()) {

2318

1

    return;

2319

1

2320

2321

93452

  Router::RouteConstSharedPtr route = std::move(vhost_route.route);

2322

2323

  // Update the cached route.

2324

93452

  setCachedRoute({route});

2325

  // Update the cached cluster info based on the new route.

2326

93452

  if (nullptr == route || nullptr == route->routeEntry()) {

2327

5481

    cached_cluster_info_ = nullptr;

2328

91106

  } else {

2329

87971

    auto* cluster = connection_manager_.cluster_manager_.getThreadLocalCluster(

2330

87971

        route->routeEntry()->clusterName());

2331

87971

    cached_cluster_info_ = (nullptr == cluster) ? nullptr : cluster->info();

2332

87971

2333

2334

  // Update route, vhost and cluster info in the filter manager's stream info.

2335

  // Now can move route here safely.

2336

93452

  filter_manager_.streamInfo().route_ = std::move(route);

2337

93452

  filter_manager_.streamInfo().vhost_ = std::move(vhost_route.vhost);

2338

93452

  filter_manager_.streamInfo().setUpstreamClusterInfo(cached_cluster_info_.value());

2339

2340

93452

  refreshTracing();

2341

93452

  refreshDurationTimeout();

2342

93452

  refreshIdleAndFlushTimeouts();

2343

93452

  refreshBufferLimit();

2344

93452

2345

2346

93452

void ConnectionManagerImpl::ActiveStream::refreshIdleAndFlushTimeouts() {

2347

93452

  if (!hasCachedRoute()) {

2348

5205

    return;

2349

5205

2350

88247

  const Router::RouteEntry* route_entry = cached_route_.value()->routeEntry();

2351

88247

  if (route_entry == nullptr) {

2352

276

    return;

2353

276

2354

2355

87971

  if (route_entry->idleTimeout().has_value()) {

2356

16193

    idle_timeout_ms_ = route_entry->idleTimeout().value();

2357

16193

    if (idle_timeout_ms_.count()) {

2358

      // If we have a route-level idle timeout but no global stream idle timeout, create a timer.

2359

16192

      if (stream_idle_timer_ == nullptr) {

2360

5

        stream_idle_timer_ = connection_manager_.dispatcher_->createScaledTimer(

2361

5

            Event::ScaledTimerType::HttpDownstreamIdleStreamTimeout,

2362

5

            [this]() -> void { onIdleTimeout(); });

2363

5

2364

16192

    } else if (stream_idle_timer_ != nullptr) {

2365

      // If we had a global stream idle timeout but the route-level idle timeout is set to zero

2366

      // (to override), we disable the idle timer.

2367

1

      stream_idle_timer_->disableTimer();

2368

1

      stream_idle_timer_ = nullptr;

2369

1

2370

16193

2371

2372

87971

  if (route_entry->flushTimeout().has_value()) {

2373

4

    response_encoder_->getStream().setFlushTimeout(route_entry->flushTimeout().value());

2374

87967

  } else if (!has_explicit_global_flush_timeout_ && route_entry->idleTimeout().has_value()) {

2375

    // If there is no route-level flush timeout, and the global flush timeout was also inherited

2376

    // from the idle timeout, also inherit the route-level idle timeout. This is for backwards

2377

    // compatibility.

2378

8

    response_encoder_->getStream().setFlushTimeout(idle_timeout_ms_);

2379

8

2380

87971

2381

2382

21

void ConnectionManagerImpl::ActiveStream::refreshAccessLogFlushTimer() {

2383

21

  if (connection_manager_.config_->accessLogFlushInterval().has_value()) {

2384

21

    access_log_flush_timer_->enableTimer(

2385

21

        connection_manager_.config_->accessLogFlushInterval().value(), this);

2386

21

2387

21

2388

2389

93452

void ConnectionManagerImpl::ActiveStream::refreshBufferLimit() {

2390

93452

  if (!hasCachedRoute()) {

2391

5205

    return;

2392

5205

2393

88247

  const Router::RouteEntry* route_entry = cached_route_.value()->routeEntry();

2394

88247

  if (route_entry == nullptr) {

2395

276

    return;

2396

276

2397

2398

87971

  const uint64_t buffer_limit = route_entry->requestBodyBufferLimit();

2399

87971

  if (buffer_limit == std::numeric_limits<uint64_t>::max()) {

2400

    // Max uint64_t means no valid limit configured.

2401

87920

    return;

2402

87920

2403

  // Only increase the buffer limit automatically. This is to ensure same

2404

  // behavior as previous logic in router filter.

2405

51

  if (buffer_limit > filter_manager_.bufferLimit()) {

2406

5

    ENVOY_STREAM_LOG(debug, "Setting new filter manager buffer limit: {}", *this, buffer_limit);

2407

5

    filter_manager_.setBufferLimit(buffer_limit);

2408

5

2409

51

2410

2411

832

void ConnectionManagerImpl::ActiveStream::clearRouteCache() {

2412

  // If the cached route is blocked then any attempt to clear it or refresh it

2413

  // will be ignored.

2414

832

  if (routeCacheBlocked()) {

2415

1

    return;

2416

1

2417

2418

831

  setCachedRoute({});

2419

831

  cached_cluster_info_ = absl::optional<Upstream::ClusterInfoConstSharedPtr>();

2420

831

2421

2422

4

void ConnectionManagerImpl::ActiveStream::refreshRouteCluster() {

2423

  // If there is no cached route, or route cache is frozen, or the request headers are not

2424

  // available, then do not refresh the route cluster.

2425

4

  if (!hasCachedRoute() || routeCacheBlocked() || request_headers_ == nullptr) {

2426

2

    return;

2427

2

2428

2

  if (const auto* entry = (*cached_route_)->routeEntry(); entry != nullptr) {

2429

    // Refresh the cluster if possible.

2430

2

    entry->refreshRouteCluster(*request_headers_, filter_manager_.streamInfo());

2431

2432

    // Refresh the cached cluster info is necessary.

2433

2

    if (!cached_cluster_info_.has_value() || cached_cluster_info_.value() == nullptr ||

2434

2

        (*cached_cluster_info_)->name() != entry->clusterName()) {

2435

2

      auto* cluster =

2436

2

          connection_manager_.cluster_manager_.getThreadLocalCluster(entry->clusterName());

2437

2

      cached_cluster_info_ = (nullptr == cluster) ? nullptr : cluster->info();

2438

2

      filter_manager_.streamInfo().setUpstreamClusterInfo(cached_cluster_info_.value());

2439

2

2440

2

2441

2

2442

2443

void ConnectionManagerImpl::ActiveStream::setCachedRoute(

2444

94283

    absl::optional<Router::RouteConstSharedPtr>&& route) {

2445

94283

  if (hasCachedRoute()) {

2446

    // The configuration of the route may be referenced by some filters.

2447

    // Cache the route to avoid it being destroyed before the stream is destroyed.

2448

587

    cleared_cached_routes_.emplace_back(std::move(cached_route_.value()));

2449

587

2450

94283

  cached_route_ = std::move(route);

2451

94283

2452

2453

47257

void ConnectionManagerImpl::ActiveStream::blockRouteCache() {

2454

47257

  route_cache_blocked_ = true;

2455

  // Clear the snapped route configuration because it is unnecessary to keep it.

2456

47257

  snapped_route_config_.reset();

2457

47257

  snapped_scoped_routes_config_.reset();

2458

47257

2459

2460

85

void ConnectionManagerImpl::ActiveStream::onRequestDataTooLarge() {

2461

85

  connection_manager_.stats_.named_.downstream_rq_too_large_.inc();

2462

85

2463

2464

void ConnectionManagerImpl::ActiveStream::recreateStream(

2465

296

    StreamInfo::FilterStateSharedPtr filter_state) {

2466

296

  ENVOY_EXECUTION_SCOPE(trackedStream(), active_span_.get());

2467

296

  ResponseEncoder* response_encoder = response_encoder_;

2468

296

  response_encoder_ = nullptr;

2469

2470

296

  Buffer::InstancePtr request_data = std::make_unique<Buffer::OwnedImpl>();

2471

296

  const auto& buffered_request_data = filter_manager_.bufferedRequestData();

2472

296

  const bool proxy_body = buffered_request_data != nullptr && buffered_request_data->length() > 0;

2473

296

  if (proxy_body) {

2474

30

    request_data->move(*buffered_request_data);

2475

30

2476

2477

296

  response_encoder->getStream().removeCallbacks(*this);

2478

2479

  // This functionally deletes the stream (via deferred delete) so do not

2480

  // reference anything beyond this point.

2481

  // Make sure to not check for deferred close as we'll be immediately creating a new stream.

2482

296

  state_.is_internally_destroyed_ = true;

2483

296

  connection_manager_.doEndStream(*this, /*check_for_deferred_close*/ false);

2484

2485

296

  RequestDecoder& new_stream = connection_manager_.newStream(*response_encoder, true);

2486

2487

  // Set the new RequestDecoder on the ResponseEncoder. Even though all of the decoder callbacks

2488

  // have already been called at this point, the encoder still needs the new decoder for deferred

2489

  // logging in some cases.

2490

  // This doesn't currently work for HTTP/1 as the H/1 ResponseEncoder doesn't hold the active

2491

  // stream's pointer to the RequestDecoder.

2492

296

  response_encoder->setRequestDecoder(new_stream);

2493

  // We don't need to copy over the old parent FilterState from the old StreamInfo if it did not

2494

  // store any objects with a LifeSpan at or above DownstreamRequest. This is to avoid unnecessary

2495

  // heap allocation.

2496

  // TODO(snowp): In the case where connection level filter state has been set on the connection

2497

  // FilterState that we inherit, we'll end up copying this every time even though we could get

2498

  // away with just resetting it to the HCM filter_state_.

2499

296

  if (filter_state->hasDataAtOrAboveLifeSpan(StreamInfo::FilterState::LifeSpan::Request)) {

2500

296

    (*connection_manager_.streams_.begin())->filter_manager_.streamInfo().filter_state_ =

2501

296

        std::make_shared<StreamInfo::FilterStateImpl>(

2502

296

            filter_state->parent(), StreamInfo::FilterState::LifeSpan::FilterChain);

2503

296

2504

2505

  // Make sure that relevant information makes it from the original stream info

2506

  // to the new one. Generally this should consist of all downstream related

2507

  // data, and not include upstream related data.

2508

296

  (*connection_manager_.streams_.begin())

2509

296

      ->filter_manager_.streamInfo()

2510

296

      .setFromForRecreateStream(filter_manager_.streamInfo());

2511

296

  new_stream.decodeHeaders(std::move(request_headers_), !proxy_body);

2512

296

  if (proxy_body) {

2513

    // This functionality is currently only used for internal redirects, which the router only

2514

    // allows if the full request has been read (end_stream = true) so we don't need to handle the

2515

    // case of upstream sending an early response mid-request.

2516

30

    new_stream.decodeData(*request_data, true);

2517

30

2518

296

2519

2520

1

Http1StreamEncoderOptionsOptRef ConnectionManagerImpl::ActiveStream::http1StreamEncoderOptions() {

2521

1

  return response_encoder_->http1StreamEncoderOptions();

2522

1

2523

2524

76

void ConnectionManagerImpl::ActiveStream::onResponseDataTooLarge() {

2525

76

  connection_manager_.stats_.named_.rs_too_large_.inc();

2526

76

2527

2528

312

void ConnectionManagerImpl::ActiveStream::resetStream(Http::StreamResetReason, absl::string_view) {

2529

312

  connection_manager_.stats_.named_.downstream_rq_tx_reset_.inc();

2530

312

  connection_manager_.doEndStream(*this);

2531

312

2532

2533

5681

bool ConnectionManagerImpl::ActiveStream::onDeferredRequestProcessing() {

2534

  // TODO(yanavlasov): Merge this with the filter manager continueIteration() method

2535

5681

  if (!state_.deferred_to_next_io_iteration_) {

2536

5327

    return false;

2537

5327

2538

354

  ENVOY_EXECUTION_SCOPE(trackedStream(), active_span_.get());

2539

354

  state_.deferred_to_next_io_iteration_ = false;

2540

354

  bool end_stream = state_.deferred_end_stream_ && deferred_data_ == nullptr &&

2541

354

                    deferred_request_trailers_ == nullptr && deferred_metadata_.empty();

2542

354

  filter_manager_.decodeHeaders(*request_headers_, end_stream);

2543

354

  if (end_stream) {

2544

60

    return true;

2545

60

2546

  // Send metadata before data, as data may have an associated end_stream.

2547

352

  while (!deferred_metadata_.empty()) {

2548

58

    MetadataMapPtr& metadata = deferred_metadata_.front();

2549

58

    filter_manager_.decodeMetadata(*metadata);

2550

58

    deferred_metadata_.pop();

2551

58

2552

  // Filter manager will return early from decodeData and decodeTrailers if

2553

  // request has completed.

2554

294

  if (deferred_data_ != nullptr) {

2555

274

    end_stream = state_.deferred_end_stream_ && deferred_request_trailers_ == nullptr;

2556

274

    filter_manager_.decodeData(*deferred_data_, end_stream);

2557

274

2558

294

  if (deferred_request_trailers_ != nullptr) {

2559

144

    request_trailers_ = std::move(deferred_request_trailers_);

2560

144

    filter_manager_.decodeTrailers(*request_trailers_);

2561

144

2562

294

  return true;

2563

354

2564

2565

90674

bool ConnectionManagerImpl::shouldDeferRequestProxyingToNextIoCycle() {

2566

  // Do not defer this stream if stream deferral is disabled

2567

90674

  if (deferred_request_processing_callback_ == nullptr) {

2568

49856

    return false;

2569

49856

2570

  // Defer this stream if there are already deferred streams, so they are not

2571

  // processed out of order

2572

40818

  if (deferred_request_processing_callback_->enabled()) {

2573

40406

    return true;

2574

40406

2575

412

  ++requests_during_dispatch_count_;

2576

412

  bool defer = requests_during_dispatch_count_ > max_requests_during_dispatch_;

2577

412

  if (defer) {

2578

328

    deferred_request_processing_callback_->scheduleCallbackNextIteration();

2579

328

2580

412

  return defer;

2581

40818

2582

2583

324

void ConnectionManagerImpl::onDeferredRequestProcessing() {

2584

324

  if (streams_.empty()) {

2585

2

    return;

2586

2

2587

322

  requests_during_dispatch_count_ = 1; // 1 stream is always let through

2588

  // Streams are inserted at the head of the list. As such process deferred

2589

  // streams in the reverse order.

2590

322

  auto reverse_iter = std::prev(streams_.end());

2591

322

  bool at_first_element = false;

2592

5681

  do {

2593

5681

    at_first_element = reverse_iter == streams_.begin();

2594

    // Move the iterator to the previous item in case the `onDeferredRequestProcessing` call removes

2595

    // the stream from the list.

2596

5681

    auto previous_element = std::prev(reverse_iter);

2597

5681

    bool was_deferred = (*reverse_iter)->onDeferredRequestProcessing();

2598

5681

    if (was_deferred && shouldDeferRequestProxyingToNextIoCycle()) {

2599

306

      break;

2600

306

2601

5375

    reverse_iter = previous_element;

2602

    // TODO(yanavlasov): see if `rend` can be used.

2603

5375

  } while (!at_first_element);

2604

322

2605

2606

} // namespace Http

2607

} // namespace Envoy