<commit_before>// Copyright 2020 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "src/envoy/token/token_subscriber.h"

#include "absl/strings/str_cat.h"
#include "api/envoy/v10/http/common/base.pb.h"
#include "envoy/http/async_client.h"
#include "envoy/http/header_map.h"
#include "source/common/common/assert.h"
#include "source/common/common/enum_to_int.h"
#include "source/common/http/message_impl.h"
#include "source/common/http/utility.h"

namespace espv2 {
namespace envoy {
namespace token {

using ::espv2::api::envoy::v10::http::common::DependencyErrorBehavior;

// Delay after a failed fetch.
constexpr std::chrono::seconds kFailedRequestRetryTime(2);

// Update the token `n` seconds before the expiration.
constexpr std::chrono::seconds kRefreshBuffer(5);

TokenSubscriber::TokenSubscriber(
    Envoy::Server::Configuration::FactoryContext& context,
    const TokenType& token_type, const std::string& token_cluster,
    const std::string& token_url, std::chrono::seconds fetch_timeout,
    DependencyErrorBehavior error_behavior, UpdateTokenCallback callback,
    TokenInfoPtr token_info)
    : context_(context),
      token_type_(token_type),
      token_cluster_(token_cluster),
      token_url_(token_url),
      fetch_timeout_(fetch_timeout),
      error_behavior_(error_behavior),
      callback_(callback),
      token_info_(std::move(token_info)),
      active_request_(nullptr),
      init_target_(nullptr) {
  debug_name_ = absl::StrCat("TokenSubscriber(", token_url_, ")");
}

void TokenSubscriber::init() {
  init_target_ = std::make_unique<Envoy::Init::TargetImpl>(
      debug_name_, [this] { refresh(); });
  refresh_timer_ = context_.mainThreadDispatcher().createTimer(
      [this]() -> void { refresh(); });

  context_.initManager().add(*init_target_);
}

TokenSubscriber::~TokenSubscriber() {
  if (active_request_) {
    active_request_->cancel();
  }
}

void TokenSubscriber::handleFailResponse() {
  active_request_ = nullptr;
  refresh_timer_->enableTimer(kFailedRequestRetryTime);

  switch (error_behavior_) {
    case DependencyErrorBehavior::ALWAYS_INIT:
      ENVOY_LOG(debug,
                "{}: Response failed, but signalling ready due to "
                "DependencyErrorBehavior config.");
      init_target_->ready();
      break;
    default:
      break;
  }
}

void TokenSubscriber::handleSuccessResponse(absl::string_view token,
                                            std::chrono::seconds expires_in) {
  active_request_ = nullptr;

  // Signal that we are ready for initialization.
  ENVOY_LOG(debug, "{}: Got token and expiry duration: {} , {} seconds",
            debug_name_, token, expires_in.count());
  callback_(token);
  init_target_->ready();

  if (expires_in <= kRefreshBuffer) {
    // Handle low expiry time by retrying immediately.
    refresh();
  } else {
    refresh_timer_->enableTimer(expires_in - kRefreshBuffer);
  }
}

void TokenSubscriber::refresh() {
  if (active_request_) {
    active_request_->cancel();
  }

  ENVOY_LOG(debug, "{}: Sending TokenSubscriber request", debug_name_);

  Envoy::Http::RequestMessagePtr message =
      token_info_->prepareRequest(token_url_);
  if (message == nullptr) {
    // Preconditions in TokenInfo are not met, not an error.
    ENVOY_LOG(warn, "{}: preconditions not met, retrying later", debug_name_);
    handleFailResponse();
    return;
  }

  const struct Envoy::Http::AsyncClient::RequestOptions options =
      Envoy::Http::AsyncClient::RequestOptions()
          .setTimeout(std::chrono::duration_cast<std::chrono::milliseconds>(
              fetch_timeout_))
          // Metadata server rejects X-Forwarded-For requests.
          // https://cloud.google.com/compute/docs/storing-retrieving-metadata#x-forwarded-for_header
          .setSendXff(false);

  const auto thread_local_cluster =
      context_.clusterManager().getThreadLocalCluster(token_cluster_);
  if (thread_local_cluster) {
    active_request_ = thread_local_cluster->httpAsyncClient().send(
        std::move(message), *this, options);
  }
}

void TokenSubscriber::processResponse(
    Envoy::Http::ResponseMessagePtr&& response) {
  try {
    const uint64_t status_code =
        Envoy::Http::Utility::getResponseStatus(response->headers());

    if (status_code != Envoy::enumToInt(Envoy::Http::Code::OK)) {
      ENVOY_LOG(error, "{}: failed: {}", debug_name_, status_code);
      handleFailResponse();
      return;
    }
  } catch (const Envoy::EnvoyException& e) {
    // This occurs if the status header is missing.
    // Catch the exception to prevent unwinding and skipping cleanup.<commit_msg>Fix service_management_config integration test (#677)<commit_after>// Copyright 2020 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include "src/envoy/token/token_subscriber.h"

#include "absl/strings/str_cat.h"
#include "api/envoy/v10/http/common/base.pb.h"
#include "envoy/http/async_client.h"
#include "envoy/http/header_map.h"
#include "source/common/common/assert.h"
#include "source/common/common/enum_to_int.h"
#include "source/common/http/message_impl.h"
#include "source/common/http/utility.h"

namespace espv2 {
namespace envoy {
namespace token {

using ::espv2::api::envoy::v10::http::common::DependencyErrorBehavior;

// Delay after a failed fetch.
constexpr std::chrono::seconds kFailedRequestRetryTime(2);

// Update the token `n` seconds before the expiration.
constexpr std::chrono::seconds kRefreshBuffer(5);

TokenSubscriber::TokenSubscriber(
    Envoy::Server::Configuration::FactoryContext& context,
    const TokenType& token_type, const std::string& token_cluster,
    const std::string& token_url, std::chrono::seconds fetch_timeout,
    DependencyErrorBehavior error_behavior, UpdateTokenCallback callback,
    TokenInfoPtr token_info)
    : context_(context),
      token_type_(token_type),
      token_cluster_(token_cluster),
      token_url_(token_url),
      fetch_timeout_(fetch_timeout),
      error_behavior_(error_behavior),
      callback_(callback),
      token_info_(std::move(token_info)),
      active_request_(nullptr),
      init_target_(nullptr) {
  debug_name_ = absl::StrCat("TokenSubscriber(", token_url_, ")");
}

void TokenSubscriber::init() {
  init_target_ = std::make_unique<Envoy::Init::TargetImpl>(
      debug_name_, [this] { refresh(); });
  refresh_timer_ = context_.mainThreadDispatcher().createTimer(
      [this]() -> void { refresh(); });

  context_.initManager().add(*init_target_);
}

TokenSubscriber::~TokenSubscriber() {
  if (active_request_) {
    active_request_->cancel();
  }
}

void TokenSubscriber::handleFailResponse() {
  active_request_ = nullptr;
  refresh_timer_->enableTimer(kFailedRequestRetryTime);

  switch (error_behavior_) {
    case DependencyErrorBehavior::ALWAYS_INIT:
      ENVOY_LOG(debug,
                "{}: Response failed, but signalling ready due to "
                "DependencyErrorBehavior config.");
      init_target_->ready();
      break;
    default:
      break;
  }
}

void TokenSubscriber::handleSuccessResponse(absl::string_view token,
                                            std::chrono::seconds expires_in) {
  active_request_ = nullptr;

  // Signal that we are ready for initialization.
  ENVOY_LOG(debug, "{}: Got token and expiry duration: {} , {} seconds",
            debug_name_, token, expires_in.count());
  callback_(token);
  init_target_->ready();

  if (expires_in <= kRefreshBuffer) {
    // Handle low expiry time by retrying immediately.
    refresh();
  } else {
    refresh_timer_->enableTimer(expires_in - kRefreshBuffer);
  }
}

void TokenSubscriber::refresh() {
  if (active_request_) {
    active_request_->cancel();
  }

  ENVOY_LOG(debug, "{}: Sending TokenSubscriber request", debug_name_);

  Envoy::Http::RequestMessagePtr message =
      token_info_->prepareRequest(token_url_);
  if (message == nullptr) {
    // Preconditions in TokenInfo are not met, not an error.
    ENVOY_LOG(warn, "{}: preconditions not met, retrying later", debug_name_);
    handleFailResponse();
    return;
  }

  const struct Envoy::Http::AsyncClient::RequestOptions options =
      Envoy::Http::AsyncClient::RequestOptions()
          .setTimeout(std::chrono::duration_cast<std::chrono::milliseconds>(
              fetch_timeout_))
          // Metadata server rejects X-Forwarded-For requests.
          // https://cloud.google.com/compute/docs/storing-retrieving-metadata#x-forwarded-for_header
          .setSendXff(false);

  const auto thread_local_cluster =
      context_.clusterManager().getThreadLocalCluster(token_cluster_);
  if (thread_local_cluster) {
    active_request_ = thread_local_cluster->httpAsyncClient().send(
        std::move(message), *this, options);
  } else {
    // This code doesn't handle the wrong cluster name case.
    // Assume it is due to the cluster is not ready.
    ENVOY_LOG(warn, "{}: the cluster {} is not ready.", debug_name_, token_cluster_);
    handleFailResponse();
  }
}

void TokenSubscriber::processResponse(
    Envoy::Http::ResponseMessagePtr&& response) {
  try {
    const uint64_t status_code =
        Envoy::Http::Utility::getResponseStatus(response->headers());

    if (status_code != Envoy::enumToInt(Envoy::Http::Code::OK)) {
      ENVOY_LOG(error, "{}: failed: {}", debug_name_, status_code);
      handleFailResponse();
      return;
    }
  } catch (const Envoy::EnvoyException& e) {
    // This occurs if the status header is missing.
    // Catch the exception to prevent unwinding and skipping cleanup.
{'fraction_non_alphanumeric': 0.12155376598768357, 'fraction_numerical': 0.003694931312174325, 'mean_word_length': 3.9863013698630136, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 27, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2630529', 'n_tokens_mistral': 3131, 'n_tokens_neox': 3067, 'n_words': 710}