Istio-proxy相关概念

istio-proxy

   Istio代理是可在客户端和服务器端使用的微服务代理，并形成微服务网格。代理支持大量功能。

 客户端功能：

• 发现和负载平衡。代理可以使用几个标准的服务发现和负载平衡API，以有效地将流量分配给服务。
• 凭证注入。代理可以通过连接隧道或特定于协议的机制（例如HTTP请求的JWT令牌）注入客户端身份。
• 连接管理。代理管理与服务的连接，处理运行状况检查，重试，故障转移和流控制。
• 监控和记录。代理可以报告客户端指标并记录到混合器。

服务器端功能：

• 速率限制和流量控制。代理可以防止后端系统过载，并提供客户端感知的速率限制。
• 协议翻译。代理是gRPC网关，提供JSON-REST和gRPC之间的转换。
• 认证与授权。代理支持多种身份验证机制，并且可以使用客户端身份通过混合器执行授权检查。
• 监控和记录。代理可以报告服务器端指标并记录到混合器。

istio proxy和Envoy的关系

istio proxy这个项目工程既包含引用了Envoy的源码，还在此基础上自己做了扩展，这个扩展是通过Envoy filter（过滤器）的形式来提供，这样的话就可以使得proxy代理将策略执行决策委托给Mixer，这样就解释了为什么Mixer可以被设计为提供策略和遥测的组件，Mixer->istio proxy->Envoy这种形式来控制。这样通过这个方式就能：

• 使用到Envoy的全部功能
• 基于Envoy做扩展，结合istio本身做处理

编译环境

centos7.x ，首先参考Bazel的官方文档安装Bazel，并且需要安装gcc等相关工具。

git clone https://github.com/istio/proxy.git
cd proxy
make build_envoy

项目主要目录如下：

├── "BUILD"
├── "Makefile"
├── "WORKSPACE"
├── src
│   ├── envoy                               -- envoy filter 插件源码
│   │   ├── alts
│   │   │   ├── *.cc
│   │   │   ├── *.h
│   │   │   └── "BUILD"
│   │   ├── "BUILD"
│   │   ├── http
│   │   │   ├── authn                     --认证 filte
│   │   │   │   ├── *.cc
│   │   │   │   ├── *.h
│   │   │   │   └── "BUILD"
│   │   │   ├── jwt_auth                    --jwt 认证 filter
│   │   │   │   ├── *.cc
│   │   │   │   ├── *.h
│   │   │   │   └── "BUILD"
│   │   │   └── mixer                      --mixer filter，实现metrics上报，Quota(Rate Limiting (处理http协议) 
│   │   │       ├── *.cc
│   │   │       ├── *.h
│   │   │       └── "BUILD"
│   │   ├── tcp
│   │   │   └── mixer                      --mixer filter(处理tcp协议)
│   │   │       ├── *.cc
│   │   │       ├── *.h
│   │   │       └── "BUILD"
│   │   └── utils
│   │       ├── *.cc
│   │       ├── *.h
│   │       └── "BUILD"
│   └── istio
│       └── **
├── test
│   └── **
└── tools
    └── **

makefile文件的关键信息:其中//src/envoy:envoy为bazel的语法，我们根据路径查询对应的内容；

build_envoy:
   export PATH=$(PATH) CC=$(CC) CXX=$(CXX) && bazel $(BAZEL_STARTUP_ARGS) build $(BAZEL_BUILD_ARGS) $(BAZEL_CONFIG_REL) //src/envoy:envoy

/src/envops/BUILD文件关键内容为：cc_binary表明该target对应的是c++二进制文件路径，其中deps部分是其依赖的其他target。前13个target都是本地依赖，对应到源码目录中的其他子目录下的BUILD文件，其中最后一个比较特殊，是一个外部依赖，该外部库为envoy，因此proxy会make时音容envoy的源码，也算是扩展吧。

envoy_cc_binary(
    name = "envoy",
    repository = "@envoy",
    visibility = ["//visibility:public"],
    deps = [
        // 对应本地文件
        "//extensions/access_log_policy:access_log_policy_lib",
        "//extensions/metadata_exchange:metadata_exchange_lib",
        "//extensions/stackdriver:stackdriver_plugin",
        "//extensions/stats:stats_plugin",
        "//src/envoy/http/alpn:config_lib",
        "//src/envoy/http/authn:filter_lib",
        "//src/envoy/http/jwt_auth:http_filter_factory",
        "//src/envoy/http/mixer:filter_lib",
        "//src/envoy/tcp/forward_downstream_sni:config_lib",
        "//src/envoy/tcp/metadata_exchange:config_lib",
        "//src/envoy/tcp/mixer:filter_lib",
        "//src/envoy/tcp/sni_verifier:config_lib",
        "//src/envoy/tcp/tcp_cluster_rewrite:config_lib",
        // 对应外部的envoy
        "@envoy//source/exe:envoy_main_entry_lib",
    ],
)

外部库定义在根目录下的workspace中，envoy的相关内容如图，在执行过程中，根据URL下载编译指定资源。

http_archive(
    name = "envoy",
    sha256 = ENVOY_SHA256,
    strip_prefix = "envoy-wasm-" + ENVOY_SHA,
    url = "https://github.com/envoyproxy/envoy-wasm/archive/" + ENVOY_SHA + ".tar.gz",
)

编译过程中的依赖关系如下图所示：

从istio-proxy项目中的Envoy BUILD中可以知道，这里会编译出name = "Envoy"的二进制程序，然后start_Envoy会启动Envoy，同时会根据一些默认参数和配置文件模板生成一个全新的配置文件，然后运行。一些关键参数如Envoy二进制的路径和配置路径、监听的端口、mixer的地址如下：

Envoy相关部分

istio-proxy源码中提供了envoy.conf.template 通用配置模板，这个模板文件最终会生成一个envoy的配置文件，然后envoy启动的时候指定运行。模板配置文件中已经配置好了Mixer相关的参数如mixer_server，这个Mixer对于Envoy来说就是一个cluster，因此是在cluster_manager里面进行管理配置。

然后Envoy的官方文档Listener discovery service (LDS)一文中有说明静态的Listener文件配置是无法通过LDS API进行修改或删除的，因此静态配置会一直生效，istio-proxy源码中则提供了Envoy的静态配置文件envoy_lds.conf 静态listeners配置

Filter相关

官方文档中描述的三个filtuers相关概念；Istio-proxy实现了Network::ReadFilter 和 Network::WriteFilter 过滤器，这样就可以通过filter API 绑定到Listener，然后当有数据读or写的时候调用到对应的filter了，这样数据流就从Envoy本身转到了filter中。

// path:/src/envoy/tcp/mixer/filter.h
// Network::ReadFilter
Network::FilterStatus onData(Buffer::Instance &data, bool) override;

// Network::WriteFilter
Network::FilterStatus onWrite(Buffer::Instance &data, bool) override;

然后network filter再转到Envoy 的 http filter，根据官方文档HTTP filters的介绍也有两种Filters，为Decoder和Encoder，然后在istio-proxy源码中实现了 Http::StreamDecoderFilter和Http::StreamEncoderFilter这两个Filter，这样的话整个流程就串起来了。

 

// path:/src/envoy/
//Http::StreamDecoderFilter
FilterHeadersStatus decodeHeaders(HeaderMap& headers, bool) override;
FilterDataStatus decodeData(Buffer::Instance& data, bool end_stream) override;

// Http::StreamEncoderFilter
FilterHeadersStatus encode100ContinueHeaders(HeaderMap&) override {
  return FilterHeadersStatus::Continue;
}
FilterHeadersStatus encodeHeaders(HeaderMap& headers, bool) override;
FilterDataStatus encodeData(Buffer::Instance&, bool) override {
  return FilterDataStatus::Continue;
}

Mixer client

原有mixer client仓库是独立的，现在已经整合到了istio-proxy的代码仓库中，这样就可以很方便的在Sidecar Envoy代理中实现：

• 和mixer server交互
• 添加一级缓存
• 前置检查
• 后置批量上报
• 策略控制
• 属性、字段转换和传递

对此Istio-proxy的理解，这个Envoy的扩展，就是在Envoy基础上，增加了一些Filter，然后通过这些个Filter在利用Mixer Client和 Mixer Server进行通信，这样就可以在proxy代理中：

• 流量代理
• 策略控制
• 遥测代理

Envoy详解

Envoy是一个高性能的C++写的proxy转发器，那Envoy如何转发请求呢？需要定一些规则，然后按照这些规则进行转发。

规则可以是静态的，放在配置文件中的，启动的时候加载，要想重新加载，一般需要重新启动，但是Envoy支持热加载和热重启，一定程度上缓解了这个问题。

当然最好的方式是规则设置为动态的，放在统一的地方维护，这个统一的地方在Envoy眼中看来称为Discovery Service，过一段时间去这里拿一下配置，就修改了转发策略。

无论是静态的，还是动态的，在配置里面往往会配置四个东西。

• listener，也即envoy既然是proxy，专门做转发，就得监听一个端口，接入请求，然后才能够根据策略转发，这个监听的端口称为listener
• endpoint，是目标的ip地址和端口，这个是proxy最终将请求转发到的地方。

• cluster，一个cluster是具有完全相同行为的多个endpoint，也即如果有三个容器在运行，就会有三个IP和端口，但是部署的是完全相同的三个服务，他们组成一个Cluster，从cluster到endpoint的过程称为负载均衡，可以轮询等。

• route，有时候多个cluster具有类似的功能，但是是不同的版本号，可以通过route规则，选择将请求路由到某一个版本号，也即某一个cluster。
• 静态配置表

 

Envoy启动过程解析

下载Envoy源码，启动入口函数在source/exe/main.cc

std::unique_ptr<Envoy::MainCommon> main_common;
// 声明并初始化Envoy::MainCommon实例为main_common
try {
  main_common = std::make_unique<Envoy::MainCommon>(argc, argv);
} catch (const Envoy::NoServingException& e) {
  return EXIT_SUCCESS;
} catch (const Envoy::MalformedArgvException& e) {
  std::cerr << e.what() << std::endl;
  return EXIT_FAILURE;
} catch (const Envoy::EnvoyException& e) {
  std::cerr << e.what() << std::endl;
  return EXIT_FAILURE;
}
// 声明并初始化Envoy::MainCommon实例为main_common，执行main_common->run启动Server
return main_common->run() ? EXIT_SUCCESS : EXIT_FAILURE;

MainCommon关键代码

class MainCommon {
public:
  MainCommon(int argc, const char* const* argv);
  // 启动函数
  bool run() { return base_.run(); }
  .............
  Server::Instance* server() { return base_.server(); }
  MainCommonBase base_;
};

MainCommonBase.cc关键代码

 

// main_common.cc
int main_common(OptionsImpl& options) {
  try {
      // 生成maincommonbase，在里面会做server instance的初始化
    MainCommonBase main_common(options);
    return main_common.run() ? EXIT_SUCCESS : EXIT_FAILURE;
  } catch (EnvoyException& e) {
    return EXIT_FAILURE;
  }
  return EXIT_SUCCESS;
}

MainCommonBase::MainCommonBase(OptionsImpl& options) : options_(options) {
    ......
    // 可以看到，MainCommon将会初始化Instance，即一个服务的实例，于是，InstanceImpl进行初始化
    server_.reset(new Server::InstanceImpl(
        options_, local_address, default_test_hooks_, *restarter_, *stats_store_, access_log_lock,
        component_factory_, std::make_unique<Runtime::RandomGeneratorImpl>(), *tls_));
    ......
}

Instance会启动初始化，在初始化核心函数中，将会进行listenerConfig的全面注册

 

// server.cc
InstanceImpl::InstanceImpl(Options& options, Network::Address::InstanceConstSharedPtr local_address,
                           TestHooks& hooks, HotRestart& restarter, Stats::StoreRoot& store,
                           Thread::BasicLockable& access_log_lock,
                           ComponentFactory& component_factory,
                           Runtime::RandomGeneratorPtr&& random_generator,
                           ThreadLocal::Instance& tls) {
    ......  
    initialize(options, local_address, component_factory);
    ......
}

void InstanceImpl::initialize(Options& options,
                              Network::Address::InstanceConstSharedPtr local_address,
                              ComponentFactory& component_factory) {
    ...
    // Handle configuration that needs to take place prior to the main configuration load.
    InstanceUtil::loadBootstrapConfig(bootstrap_, options,
                 messageValidationContext().staticValidationVisitor(), *api_);
    .......
    // 初始化ListenerManager
    listener_manager_.reset(new ListenerManagerImpl(
      *this, listener_component_factory_, worker_factory_, ProdSystemTimeSource::instance_));

    // 会初始化
    main_config->initialize(bootstrap_, *this, *cluster_manager_factory_);

    ...
}
// 通过loadFromFile和loadFromYaml读取配置文件路径下的配置，并完成参数校验。
InstanceUtil::loadBootstrapConfig(envoy::config::bootstrap::v2::Bootstrap& bootstrap,
                                  Options& options) {
  try {
      // 根据配置信息查找对应配置文件
    if (!options.configPath().empty()) {
      MessageUtil::loadFromFile(options.configPath(), bootstrap);
    }
    if (!options.configYaml().empty()) {
      envoy::config::bootstrap::v2::Bootstrap bootstrap_override;
      MessageUtil::loadFromYaml(options.configYaml(), bootstrap_override);
      bootstrap.MergeFrom(bootstrap_override);
    }
    MessageUtil::validate(bootstrap);
    return BootstrapVersion::V2;
  } catch (const EnvoyException& e) {
    if (options.v2ConfigOnly()) {
      throw;
    }
    // TODO(htuch): When v1 is deprecated, make this a warning encouraging config upgrade.
    ENVOY_LOG(debug, "Unable to initialize config as v2, will retry as v1: {}", e.what());
  }
  if (!options.configYaml().empty()) {
    throw EnvoyException("V1 config (detected) with --config-yaml is not supported");
  }
  Json::ObjectSharedPtr config_json = Json::Factory::loadFromFile(options.configPath());
  Config::BootstrapJson::translateBootstrap(*config_json, bootstrap);
  MessageUtil::validate(bootstrap);
  return BootstrapVersion::V1;
}

void MainImpl::initialize(const envoy::config::bootstrap::v2::Bootstrap& bootstrap,
                          Instance& server,
                          Upstream::ClusterManagerFactory& cluster_manager_factory) {
  ......
  const auto& listeners = bootstrap.static_resources().listeners();
  ENVOY_LOG(info, "loading {} listener(s)", listeners.size());
  // 从bootstrap配置（yaml文件）中提取listener配置，并依次进行添加操作。
  for (ssize_t i = 0; i < listeners.size(); i++) {
    ENVOY_LOG(debug, "listener #{}:", i);
    server.listenerManager().addOrUpdateListener(listeners[i], "", false);
  }
  ...... 

回到MainCommonBase.run方法

bool MainCommonBase::run() {
  switch (options_.mode()) {
  case Server::Mode::Serve:
    server_->run();
    return true;
  case Server::Mode::Validate: {
    auto local_address = Network::Utility::getLocalAddress(options_.localAddressIpVersion());
    return Server::validateConfig(options_, local_address, component_factory_, thread_factory_,
                                  file_system_);
  }
  case Server::Mode::InitOnly:
    PERF_DUMP();
    return true;
  }
  NOT_REACHED_GCOVR_EXCL_LINE;
}

调用到InstanceImpl的run方法

 

void InstanceImpl::run() {
  // startWorker即会进行eventloop
  RunHelper helper(*dispatcher_, clusterManager(), restarter_, access_log_manager_, init_manager_,
                   [this]() -> void { startWorkers(); });
  ......
}

void InstanceImpl::startWorkers() {
  listener_manager_->startWorkers(*guard_dog_);
  ......
}

InstanceImpl::startWorkers()方法解析

void ListenerManagerImpl::startWorkers(GuardDog& guard_dog) {
  ENVOY_LOG(info, "all dependencies initialized. starting workers");
  ASSERT(!workers_started_);
  workers_started_ = true;
  for (const auto& worker : workers_) {
    ASSERT(warming_listeners_.empty());
    for (const auto& listener : active_listeners_) {
      // 此处即会将所有listener绑定到所有worker身上。worker即服务的并发线程数。
      addListenerToWorker(*worker, *listener);
    }
    worker->start(guard_dog);
  }
}

void WorkerImpl::addListener(Network::ListenerConfig& listener, AddListenerCompletion completion) {
   ......
   handler_->addListener(listener);
   ......
}

进一步看addListener()方法

 

void ConnectionHandlerImpl::addListener(Network::ListenerConfig& config) {
  // 生成ActiveListener
  ActiveListenerPtr l(new ActiveListener(*this, config));
  listeners_.emplace_back(config.socket().localAddress(), std::move(l));
}

ConnectionHandlerImpl::ActiveListener::ActiveListener(ConnectionHandlerImpl& parent,
                                                      Network::ListenerConfig& config)
    : ActiveListener(
          parent,
          // 可以看到，在ActiveListener初始化过程中，将进行真正Listener的初始化。
          parent.dispatcher_.createListener(config.socket(), *this, config.bindToPort(),
                                            config.handOffRestoredDestinationConnections()),
          config) {}

ListenerImpl::ListenerImpl(Event::DispatcherImpl& dispatcher, Socket& socket, ListenerCallbacks& cb,
                           bool bind_to_port, bool hand_off_restored_destination_connections)
    : local_address_(nullptr), cb_(cb),
      hand_off_restored_destination_connections_(hand_off_restored_destination_connections),
    ......
    // 通过libevent的`evconnlistener_new`实现对指定监听fd的新连接事件的回调处理。
    listener_.reset(
        evconnlistener_new(&dispatcher.base(), listenCallback, this, 0, -1, socket.fd()));
    ......
}
Listener对新连接设置回调函数 & Listener Filter创建
void ListenerImpl::listenCallback(evconnlistener*, evutil_socket_t fd, sockaddr* remote_addr,
                                  int remote_addr_len, void* arg) {
  ......
  // 此处的fd已经不是listenfd，已经是该新连接的connfd。
  listener->cb_.onAccept(std::make_unique<AcceptedSocketImpl>(fd, local_address, remote_address),
                         listener->hand_off_restored_destination_connections_);
  ......
}

// 回调时候主要做两件事情，
// 1. 构建出对应的Listener Accept Filter
// 2. 构建出ServerConnection
void ConnectionHandlerImpl::ActiveListener::onAccept(
    Network::ConnectionSocketPtr&& socket, bool hand_off_restored_destination_connections) {
  ......
  auto active_socket = std::make_unique<ActiveSocket>(*this, std::move(socket),
                                                 hand_off_restored_destination_connections);
  // 构建对应的Filter
  config_.filterChainFactory().createListenerFilterChain(*active_socket);
  active_socket->continueFilterChain(true);
  ......
}

void ConnectionHandlerImpl::ActiveSocket::continueFilterChain(bool success) {
   ......
   // 创建连接
   listener_.newConnection(std::move(socket_));
   ......  
}

我们接着newConnecion方法看

 

void ConnectionHandlerImpl::ActiveListener::newConnection(Network::ConnectionSocketPtr&& socket) {
  ......
  auto transport_socket = filter_chain->transportSocketFactory().createTransportSocket();
  // 创建ServerConnection
  Network::ConnectionPtr new_connection =
      parent_.dispatcher_.createServerConnection(std::move(socket), std::move(transport_socket));
  new_connection->setBufferLimits(config_.perConnectionBufferLimitBytes());
  // 创建真正的Read/Write Filter
 const bool empty_filter_chain = !config_.filterChainFactory().createNetworkFilterChain(
      *new_connection, filter_chain->networkFilterFactories());
  ......
}

ConnectionImpl::ConnectionImpl(Event::Dispatcher& dispatcher, ConnectionSocketPtr&& socket,
                               TransportSocketPtr&& transport_socket, bool connected)
    : transport_socket_(std::move(transport_socket)), filter_manager_(*this, *this),
      socket_(std::move(socket)), write_buffer_(dispatcher.getWatermarkFactory().create(
                                      [this]() -> void { this->onLowWatermark(); },
                                      [this]() -> void { this->onHighWatermark(); })),
      dispatcher_(dispatcher), id_(next_global_id_++) {
    // 当read/writed生成事件
  file_event_ = dispatcher_.createFileEvent(
      fd(), [this](uint32_t events) -> void { onFileEvent(events); }, Event::FileTriggerType::Edge,
      Event::FileReadyType::Read | Event::FileReadyType::Write);
}

初始化完了连接和Listener，也初始化完了Accept Listener Filter和Read/Write Listener Filter，此时，Listener的libevent事件已经准备就绪，有请求到来后，Connection的read/write事件也将被触发。此时Worker启动源码：

void WorkerImpl::start(GuardDog& guard_dog) {
  ASSERT(!thread_);
  thread_.reset(new Thread::Thread([this, &guard_dog]() -> void { threadRoutine(guard_dog); }));
}

void WorkerImpl::threadRoutine(GuardDog& guard_dog) {
  ......
  dispatcher_->run(Event::Dispatcher::RunType::Block);

  // 异常退出后做的清理操作
  guard_dog.stopWatching(watchdog);
  handler_.reset();
  tls_.shutdownThread();
  watchdog.reset();
}

void DispatcherImpl::run(RunType type) {
  // 启动libevent处理
  event_base_loop(base_.get(), type == RunType::NonBlock ? EVLOOP_NONBLOCK : 0);
}

总结

     对于istio-proxy的相关概念，envoy的启动部分源码做了学习，对于启动后的动态过程，后续学习发布。