implement healthcheck endpoint for active JWKs existence (#174)

* implement healthcheck endpoint for active JWKs existence

Signed-off-by: Shikugawa <Shikugawa@gmail.com>

* fix

Signed-off-by: Shikugawa <Shikugawa@gmail.com>

* fix

Signed-off-by: Shikugawa <Shikugawa@gmail.com>

* fix

Signed-off-by: Shikugawa <Shikugawa@gmail.com>

* fix

Signed-off-by: Shikugawa <Shikugawa@gmail.com>

* fix

Signed-off-by: Shikugawa <Shikugawa@gmail.com>

* fix

Signed-off-by: Shikugawa <Shikugawa@gmail.com>

* fix

Signed-off-by: Shikugawa <Shikugawa@gmail.com>

* fix

Signed-off-by: Shikugawa <Shikugawa@gmail.com>

* fix

Signed-off-by: Shikugawa <Shikugawa@gmail.com>

* fix

Signed-off-by: Shikugawa <Shikugawa@gmail.com>

* fix

Signed-off-by: Shikugawa <Shikugawa@gmail.com>

bookinfo-example/authservice/templates/deployment.yaml

@@ -29,6 +29,9 @@ spec:
         volumeMounts:
           - name: authservice-config # mount the volume containing the authservice ConfigMap
             mountPath: /etc/authservice
+        readinessProbe:
+          path: /healthz
+          port: 10004
 ---
 apiVersion: v1
 kind: Service

src/common/http/http.cc

@@ -12,6 +12,7 @@
 #include "absl/strings/match.h"
 #include "absl/strings/str_join.h"
 #include "absl/strings/str_split.h"
+#include "boost/beast/core/tcp_stream.hpp"
 #include "spdlog/spdlog.h"
 
 namespace beast = boost::beast;    // from <boost/beast.hpp>
@@ -565,7 +566,6 @@ response_t HttpImpl::Get(
             absl::StrCat("http connect failed with status: ",
                          http_connect_res.result_int()));
       }
-
     } else {
       spdlog::info("{}: opening connection to {}:{}", __func__,
                    parsed_uri.GetHost(), parsed_uri.GetPort());
@@ -575,6 +575,7 @@ response_t HttpImpl::Get(
     }
 
     stream.async_handshake(ssl::stream_base::client, yield);
+
     // Set up an HTTP POST request message
     beast::http::request<beast::http::string_body> req{
         beast::http::verb::get, parsed_uri.GetPathQueryFragment(), version};
@@ -621,6 +622,53 @@ response_t HttpImpl::Get(
   }
 }
 
+response_t HttpImpl::SimpleGet(
+    absl::string_view uri,
+    const std::map<absl::string_view, absl::string_view> &headers,
+    absl::string_view body, boost::asio::io_context &ioc,
+    boost::asio::yield_context yield) const {
+  spdlog::trace("{}", __func__);
+  try {
+    int version = 11;
+
+    auto parsed_uri = http::Uri(uri);
+
+    tcp::resolver resolver(ioc);
+    beast::tcp_stream stream(ioc);
+
+    spdlog::info("{}: opening connection to {}:{}", __func__,
+                 parsed_uri.GetHost(), parsed_uri.GetPort());
+    const auto results = resolver.async_resolve(
+        parsed_uri.GetHost(), std::to_string(parsed_uri.GetPort()), yield);
+    beast::get_lowest_layer(stream).async_connect(results, yield);
+
+    // Set up an HTTP simple get request message
+    beast::http::request<beast::http::string_body> req{
+        beast::http::verb::get, parsed_uri.GetPathQueryFragment(), version};
+    req.set(beast::http::field::host, parsed_uri.GetHost());
+    for (auto header : headers) {
+      req.set(boost::beast::string_view(header.first.data()),
+              boost::beast::string_view(header.second.data()));
+    }
+    auto &req_body = req.body();
+    req_body.reserve(body.size());
+    req_body.append(body.begin(), body.end());
+    req.prepare_payload();
+    // Send the HTTP request to the remote host
+    beast::http::async_write(stream, req, yield);
+
+    // Read response
+    beast::flat_buffer buffer;
+    response_t res(new beast::http::response<beast::http::string_body>);
+    beast::http::async_read(stream, buffer, *res, yield);
+    return res;
+    // If we get here then the connection is closed gracefully
+  } catch (std::exception const &e) {
+    spdlog::error("{}: unexpected exception: {}", __func__, e.what());
+    return response_t();
+  }
+}
+
 }  // namespace http
 }  // namespace common
 }  // namespace authservice

src/common/http/http.h

@@ -171,6 +171,8 @@ class Http {
    */
   virtual ~Http() = default;
 
+  // TODO(shikugawa): add transport socket abstraction to enable raw or tls
+  // socket switching easiliy.
   /** @brief Asynchronously send a Post http message with a certificate
    * authority. To be used inside a Boost co-routine.
    * @param endpoint the endpoint to call
@@ -200,6 +202,19 @@ class Http {
       absl::string_view body, absl::string_view ca_cert,
       absl::string_view proxy_uri, boost::asio::io_context &ioc,
       boost::asio::yield_context yield) const = 0;
+
+  /** @brief Asynchronously send a non-SSL Get http message with a certificate
+   * authority. To be used inside a Boost co-routine.
+   * @param endpoint the endpoint to call
+   * @param headers the http headers
+   * @param body the http request body
+   * @return http response.
+   */
+  virtual response_t SimpleGet(
+      absl::string_view uri,
+      const std::map<absl::string_view, absl::string_view> &headers,
+      absl::string_view body, boost::asio::io_context &ioc,
+      boost::asio::yield_context yield) const = 0;
 };
 
 /**
@@ -218,6 +233,12 @@ class HttpImpl : public Http {
                  absl::string_view body, absl::string_view ca_cert,
                  absl::string_view proxy_uri, boost::asio::io_context &ioc,
                  boost::asio::yield_context yield) const override;
+
+  response_t SimpleGet(
+      absl::string_view uri,
+      const std::map<absl::string_view, absl::string_view> &headers,
+      absl::string_view body, boost::asio::io_context &ioc,
+      boost::asio::yield_context yield) const override;
 };
 
 }  // namespace http

src/filters/filter_chain.cc

@@ -71,7 +71,7 @@ FilterChainImpl::FilterChainImpl(boost::asio::io_context& ioc,
     }
 
     jwks_resolver_cache_ =
-        std::make_unique<oidc::JwksResolverCache>(oidc_filter->oidc(), ioc);
+        std::make_unique<oidc::JwksResolverCacheImpl>(oidc_filter->oidc(), ioc);
   }
 
   // Create filter chain factory
@@ -131,5 +131,15 @@ void FilterChainImpl::DoPeriodicCleanup() {
   }
 }
 
+bool FilterChainImpl::jwksActive() const {
+  const auto resolver = jwks_resolver_cache_->getResolver();
+  return resolver != nullptr && resolver->jwks() != nullptr;
+}
+
+void FilterChainImpl::setJwksResolverCacheForTest(
+    oidc::JwksResolverCachePtr jwks_resolver_cache) {
+  jwks_resolver_cache_ = jwks_resolver_cache;
+}
+
 }  // namespace filters
 }  // namespace authservice

src/filters/filter_chain.h

@@ -50,6 +50,18 @@ class FilterChain {
    * sessions and any other resources.
    */
   virtual void DoPeriodicCleanup() = 0;
+
+  /**
+   * Return true if JWKs on each filters are active.
+   */
+  virtual bool jwksActive() const = 0;
+
+  /**
+   * Replace existing jwks resolver cache.
+   * This will be utilized in unit test for dependency injection.
+   */
+  virtual void setJwksResolverCacheForTest(
+      oidc::JwksResolverCachePtr jwks_resolver_cache) = 0;
 };
 
 class FilterChainImpl : public FilterChain {
@@ -72,6 +84,11 @@ class FilterChainImpl : public FilterChain {
   std::unique_ptr<Filter> New() override;
 
   virtual void DoPeriodicCleanup() override;
+
+  bool jwksActive() const override;
+
+  void setJwksResolverCacheForTest(
+      oidc::JwksResolverCachePtr jwks_resolver_cache) override;
 };
 
 }  // namespace filters

src/filters/oidc/jwks_resolver.cc

@@ -48,8 +48,6 @@ void DynamicJwksResolverImpl::JwksFetcher::request(
       }
     }
 
-    // TODO(shikugawa): add healthcheck endpoint for gRPC that works as JWKs are
-    // not empty.
     if (parent_->jwks() == nullptr) {
       // On Kubernetes, depending on the timing of the Pod startup,
       // the first egress HTTP/HTTPS request may fail. This problem

src/filters/oidc/jwks_resolver.h

@@ -122,8 +122,15 @@ class DynamicJwksResolverImpl : public JwksResolver {
 
 class JwksResolverCache {
  public:
-  JwksResolverCache(const config::oidc::OIDCConfig& config,
-                    boost::asio::io_context& ioc)
+  virtual ~JwksResolverCache() = default;
+
+  virtual JwksResolverPtr getResolver() = 0;
+};
+
+class JwksResolverCacheImpl final : public JwksResolverCache {
+ public:
+  JwksResolverCacheImpl(const config::oidc::OIDCConfig& config,
+                        boost::asio::io_context& ioc)
       : config_(config) {
     switch (config_.jwks_config_case()) {
       case config::oidc::OIDCConfig::kJwks:

src/filters/oidc/redis_session_store.h

@@ -1,8 +1,6 @@
 #ifndef AUTHSERVICE_REDIS_SESSION_STORE_H
 #define AUTHSERVICE_REDIS_SESSION_STORE_H
 
-#include <cstdint>
-
 #include "config/oidc/config.pb.h"
 #include "src/common/utilities/synchronized.h"
 #include "src/common/utilities/time_service.h"

src/service/BUILD

@@ -1,5 +1,19 @@
 package(default_visibility = ["//visibility:public"])
 
+cc_library(
+    name = "healthcheck_http_server_lib",
+    hdrs = [
+        "healthcheck_http_server.h"
+    ],
+    srcs = [
+        "healthcheck_http_server.cc",
+    ],
+    deps = [
+        "//src/filters:filter_chain",
+        "@boost//:all"
+    ]
+)
+
 cc_library(
     name = "serviceimpl",
     srcs = [
@@ -9,6 +23,7 @@ cc_library(
         "async_service_impl.h",
     ],
     deps = [
+        ":healthcheck_http_server_lib",
         "//config:config_cc",
         "//src/config",
         "//src/filters:filter_chain",

src/service/async_service_impl.cc

@@ -7,12 +7,18 @@
 #include <memory>
 #include <stdexcept>
 
+#include "boost/asio/io_context.hpp"
+#include "boost/thread/detail/thread.hpp"
 #include "src/common/http/http.h"
 #include "src/config/get_config.h"
 
 namespace authservice {
 namespace service {
 
+namespace {
+constexpr uint16_t kHealthCheckServerPort = 10004;
+}
+
 ProcessingStateV2::ProcessingStateV2(
     ProcessingStateFactory &parent,
     envoy::service::auth::v2::Authorization::AsyncService &service)
@@ -170,6 +176,11 @@ void AsyncAuthServiceImpl::Run() {
     });
   }
 
+  spdlog::info("{}: Healthcheck Server listening on {}:{}", __func__,
+               config_.listen_address(), kHealthCheckServerPort);
+  health_server_ = std::make_unique<HealthcheckAsyncServer>(
+      chains_, config_.listen_address(), kHealthCheckServerPort);
+
   spdlog::info("{}: Server listening on {}", __func__, address_and_port_);
 
   try {
@@ -190,7 +201,6 @@ void AsyncAuthServiceImpl::Run() {
       if (!ok) {
         spdlog::error("{}: Unexpected error: !ok", __func__);
       }
-
       static_cast<ServiceState *>(tag)->Proceed();
     }
   } catch (const std::exception &e) {
@@ -203,6 +213,9 @@ void AsyncAuthServiceImpl::Run() {
   server_->Shutdown();
   cq_->Shutdown();
 
+  // Start shutting down health server
+  health_server_.reset();
+
   // The destructor of the completion queue will abort if there are any
   // outstanding events, so we must drain the queue before we allow that to
   // happen

src/service/async_service_impl.h

@@ -14,6 +14,7 @@
 #include "src/common/http/http.h"
 #include "src/common/utilities/trigger_rules.h"
 #include "src/filters/filter_chain.h"
+#include "src/service/healthcheck_http_server.h"
 
 namespace authservice {
 namespace service {
@@ -229,6 +230,7 @@ class AsyncAuthServiceImpl {
   envoy::service::auth::v3::Authorization::AsyncService service_;
   std::unique_ptr<grpc::ServerCompletionQueue> cq_;
   std::unique_ptr<grpc::Server> server_;
+  std::unique_ptr<HealthcheckAsyncServer> health_server_;
 
   std::shared_ptr<boost::asio::io_context> io_context_;