Feature/42 request middleware (#46)

Base.run/5 refactored to middleware. Stubs for response middleware.

.gitignore

@@ -28,3 +28,7 @@ k8s-*.tar
 
 # Ignore master version as its always pulled w/ make test/all
 test/support/swagger/master.json
+
+# Ignore dialyzer files
+/priv/plts/*.plt
+/priv/plts/*.plt.hash

.travis.yml

@@ -1,5 +1,8 @@
 language: elixir
 sudo: false
+cache:
+  directories:
+    - priv/plts
 elixir:
   - 1.8
   - 1.7

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Added
+- Request middleware support
+
 ## [0.4.0] - 2019-08-29
 
 ### Changed

Makefile

@@ -5,6 +5,9 @@ help: ## Show this help
 help:
 	@grep -E '^[\/a-zA-Z0-9._%-]+:.*?## .*$$' Makefile | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-30s\033[0m %s\n", $$1, $$2}'
 
+quality: ## Run code quality and test targets
+quality: cov lint analyze
+
 clean: ## Remove build/doc dirs
 	rm -rf _build
 	rm -rf cover
@@ -33,7 +36,7 @@ test/master: ## Run test suite against master
 	K8S_SPEC=${MASTER_SWAGGER_PATH} mix test
 
 test/all: ## Run full test suite against 1.10+
-test/all: test/1.10 test/1.11 test/1.12 test/1.13 test/1.14 test/1.15	
+test/all: test/1.10 test/1.11 test/1.12 test/1.13 test/1.14 test/1.15
 
 test/%: ## Run full test suite against a specific k8s version
 	K8S_SPEC=test/support/swagger/$*.json mix test

README.md

@@ -26,6 +26,7 @@
 * Kube config file parsing
 * Certificate and service account based auth
 * Pluggable auth providers
+* HTTP Request middleware
 * Macro free; fast compile & fast startup
 
 ## Installation

coveralls.json

@@ -13,6 +13,7 @@
     "def.+(.+\/\/.+).+do"
   ],
   "coverage_options": {
-    "treat_no_relevant_lines_as_covered": true
+    "treat_no_relevant_lines_as_covered": true,
+    "minimum_coverage": 88
   }
 }

lib/k8s/application.ex

@@ -8,7 +8,11 @@ defmodule K8s.Application do
     :ets.new(K8s.Conn, [:set, :public, :named_table])
     :ets.new(K8s.Cluster.Group, [:set, :public, :named_table])
 
-    children = [{K8s.Cluster.Registry, []}]
+    # TODO: register defaults for each cluster
+    children = [
+      {K8s.Middleware.Registry, []},
+      {K8s.Cluster.Registry, []}
+    ]
 
     opts = [strategy: :one_for_one, name: K8s.Supervisor]
     Supervisor.start_link(children, opts)

lib/k8s/client/runner/base.ex

@@ -3,11 +3,19 @@ defmodule K8s.Client.Runner.Base do
   Base HTTP processor for `K8s.Client`
   """
 
-  @type result_t :: {:ok, map() | reference()} | {:error, atom} | {:error, binary()}
+  @type result_t ::
+          {:ok, map() | reference()}
+          # | {:error, atom | binary() | K8s.Middleware.Error.t()}
+          | {:error, K8s.Middleware.Error.t()}
+          | {:error, :cluster_not_registered | :missing_required_param | :unsupported_api_version}
+          | {:error, binary()}
+
+  @typedoc "Acceptable HTTP body types"
+  @type body_t :: list(map()) | map() | binary() | nil
 
   alias K8s.Cluster
-  alias K8s.Conn.RequestOptions
   alias K8s.Operation
+  alias K8s.Middleware.Request
 
   @doc """
   Runs a `K8s.Operation`.
@@ -72,46 +80,38 @@ defmodule K8s.Client.Runner.Base do
 
   @doc """
   Run an operation and pass `opts` to HTTPoison.
+  Destructures `Operation` data and passes as the HTTP body.
 
   See `run/2`
   """
-  @spec run(Operation.t(), binary | atom, keyword()) :: result_t
+  @spec run(Operation.t(), atom, keyword()) :: result_t
   def run(%Operation{} = operation, cluster_name, opts) when is_list(opts) do
     run(operation, cluster_name, operation.data, opts)
   end
 
   @doc """
-  Run an operation with an alternative HTTP Body (map) and pass `opts` to HTTPoison.
+  Run an operation with an HTTP Body (map) and pass `opts` to HTTPoison.
   See `run/2`
   """
   @spec run(Operation.t(), atom, map(), keyword()) :: result_t
-  def run(%Operation{} = operation, cluster_name, body, opts \\ []) do
-    with {:ok, url} <- Cluster.url_for(operation, cluster_name),
-         {:ok, conn} <- Cluster.conn(cluster_name),
-         {:ok, request_options} <- RequestOptions.generate(conn),
-         {:ok, http_body} <- encode(body, operation.method) do
-      http_headers = K8s.http_provider().headers(operation.method, request_options)
-
-      http_opts_params = build_http_params(opts[:params], operation.label_selector)
-      opts_with_selector_params = Keyword.put(opts, :params, http_opts_params)
-      http_opts = Keyword.merge([ssl: request_options.ssl_options], opts_with_selector_params)
-
-      K8s.http_provider().request(
-        operation.method,
-        url,
-        http_body,
-        http_headers,
-        http_opts
-      )
+  def run(%Operation{} = operation, cluster, body, opts \\ []) do
+    with {:ok, url} <- Cluster.url_for(operation, cluster),
+         req <- new_request(cluster, url, operation, body, opts),
+         {:ok, req} <- K8s.Middleware.run(req) do
+      K8s.http_provider().request(req.method, req.url, req.body, req.headers, req.opts)
     end
   end
 
-  @spec encode(any(), atom()) :: {:ok, binary} | {:error, any}
-  def encode(body, http_method) when http_method in [:put, :patch, :post] do
-    Jason.encode(body)
-  end
+  @spec new_request(atom(), String.t(), K8s.Operation.t(), body_t, Keyword.t()) ::
+          Request.t()
+  defp new_request(cluster, url, %Operation{} = operation, body, opts) do
+    req = %Request{cluster: cluster, method: operation.method, body: body}
+    http_opts_params = build_http_params(opts[:params], operation.label_selector)
+    opts_with_selector_params = Keyword.put(opts, :params, http_opts_params)
 
-  def encode(_, _), do: {:ok, ""}
+    http_opts = Keyword.merge(req.opts, opts_with_selector_params)
+    %Request{req | opts: http_opts, url: url}
+  end
 
   @spec build_http_params(nil | keyword | map, nil | K8s.Selector.t()) :: map()
   defp build_http_params(nil, nil), do: %{}

lib/k8s/client/runner/watch.ex

@@ -78,8 +78,8 @@ defmodule K8s.Client.Runner.Watch do
         rv = parse_resource_version(payload)
         {:ok, rv}
 
-      {:error, error} ->
-        {:error, error}
+      error ->
+        error
     end
   end
 

lib/k8s/cluster.ex

@@ -18,7 +18,7 @@ defmodule K8s.Cluster do
 
   """
   @spec url_for(Operation.t(), atom) :: {:ok, binary} | {:error, atom(), binary()}
-  def url_for(%Operation{api_version: api_version, name: name, verb: verb} = operation, cluster) do
+  def url_for(%Operation{api_version: api_version, name: name, verb: _verb} = operation, cluster) do
     with {:ok, conn} <- Cluster.conn(cluster),
          {:ok, name} <- Cluster.Group.resource_name_for_kind(cluster, api_version, name),
          operation <- Map.put(operation, :name, name),

lib/k8s/cluster/registry.ex

@@ -33,6 +33,7 @@ defmodule K8s.Cluster.Registry do
   @spec add(atom(), K8s.Conn.t()) :: {:ok, atom()} | {:error, atom()}
   def add(cluster, conn) do
     with true <- :ets.insert(K8s.Conn, {cluster, conn}),
+         :ok <- K8s.Middleware.initialize(cluster),
          {:ok, resources_by_group} <- Discovery.resources_by_group(cluster) do
       K8s.Cluster.Group.insert_all(cluster, resources_by_group)
       K8s.Sys.Event.cluster_registered(%{}, %{cluster: cluster})

lib/k8s/middleware.ex

@@ -0,0 +1,59 @@
+defmodule K8s.Middleware do
+  @moduledoc "Interface for interacting with cluster middleware"
+
+  alias K8s.Middleware.{Error, Request}
+
+  @typedoc "Middleware type"
+  @type type_t :: :request | :response
+
+  @typedoc "List of middlewares"
+  @type stack_t :: list(module())
+
+  @spec defaults(K8s.Middleware.type_t()) :: stack_t
+  def defaults(:request), do: [Request.Initialize, Request.EncodeBody]
+  def defaults(:response), do: []
+
+  @doc "Initialize a clusters middleware stacks"
+  @spec initialize(atom) :: :ok
+  def initialize(cluster) do
+    K8s.Middleware.Registry.set(cluster, :request, defaults(:request))
+    K8s.Middleware.Registry.set(cluster, :response, defaults(:response))
+  end
+
+  @doc """
+  Applies middlewares registered to a `K8s.Cluster` to a `K8s.Middleware.Request`
+  """
+  @spec run(Request.t()) :: {:ok, Request.t()} | {:error, Error.t()}
+  def run(req) do
+    middlewares = K8s.Middleware.Registry.list(req.cluster, :request)
+    run(req, middlewares)
+  end
+
+  @spec run(Request.t(), list(module())) :: {:ok, Request.t()} | {:error, Error.t()}
+  def run(req, middlewares) do
+    result =
+      Enum.reduce_while(middlewares, req, fn middleware, req ->
+        case apply(middleware, :call, [req]) do
+          {:ok, updated_request} ->
+            {:cont, updated_request}
+
+          {:error, error} ->
+            {:halt, error(middleware, req, error)}
+        end
+      end)
+
+    case result do
+      %Request{} -> {:ok, result}
+      %Error{} -> {:error, result}
+    end
+  end
+
+  @spec error(module(), Request.t(), any()) :: Error.t()
+  defp error(middleware, req, error) do
+    %K8s.Middleware.Error{
+      middleware: middleware,
+      error: error,
+      request: req
+    }
+  end
+end

lib/k8s/middleware/error.ex

@@ -0,0 +1,17 @@
+defmodule K8s.Middleware.Error do
+  @moduledoc "Encapsulates middleware process errors"
+
+  @typedoc """
+  Middleware processing error
+
+  * `middleware` middleware module that caused the error
+  * `request` `K8s.Middleware.Request`
+  * `error` actual error, can be `any()` type
+  """
+  @type t :: %__MODULE__{
+          request: K8s.Middleware.Request.t() | nil,
+          middleware: module(),
+          error: any()
+        }
+  defstruct [:request, :middleware, :error]
+end

lib/k8s/middleware/registry.ex

@@ -0,0 +1,41 @@
+defmodule K8s.Middleware.Registry do
+  @moduledoc "Cluster middleware registry"
+  use Agent
+
+  @spec start_link(Keyword.t()) :: Agent.on_start()
+  def start_link(_opts) do
+    Agent.start_link(fn -> %{} end, name: __MODULE__)
+  end
+
+  @doc "Adds a middleware to the end of the middleware stack"
+  @spec add(atom, K8s.Middleware.type_t(), module()) :: :ok
+  def add(cluster, type, middleware) do
+    Agent.update(__MODULE__, fn registry ->
+      cluster_middlewares = Map.get(registry, cluster, %{})
+      middleware_list = Map.get(cluster_middlewares, type, [])
+
+      updated_middleware_list = middleware_list ++ [middleware]
+      updated_cluster_middlewares = Map.put(cluster_middlewares, type, updated_middleware_list)
+
+      put_in(registry, [cluster], updated_cluster_middlewares)
+    end)
+  end
+
+  @doc "Sets/replaces the middleware stack"
+  @spec set(atom, K8s.Middleware.type_t(), list(module())) :: :ok
+  def set(cluster, type, middlewares) do
+    Agent.update(__MODULE__, fn registry ->
+      cluster_middlewares = Map.get(registry, cluster, %{})
+      updated_cluster_middlewares = Map.put(cluster_middlewares, type, middlewares)
+
+      put_in(registry, [cluster], updated_cluster_middlewares)
+    end)
+  end
+
+  @doc "Returns middleware stack for a cluster and (request or response)"
+  @spec list(atom, K8s.Middleware.type_t()) :: K8s.Middleware.stack_t()
+  def list(cluster, type) do
+    registry = Agent.get(__MODULE__, & &1[cluster]) || %{}
+    Map.get(registry, type, [])
+  end
+end

lib/k8s/middleware/request.ex

@@ -0,0 +1,18 @@
+defmodule K8s.Middleware.Request do
+  @moduledoc "HTTP Request middleware"
+
+  @typedoc "Middleware Request type"
+  @type t :: %__MODULE__{
+          cluster: atom(),
+          method: atom(),
+          url: String.t(),
+          body: String.t() | map() | list(map()) | nil,
+          headers: Keyword.t() | nil,
+          opts: Keyword.t() | nil
+        }
+
+  defstruct cluster: nil, method: nil, url: nil, body: nil, headers: [], opts: []
+
+  @doc "Request middleware callback"
+  @callback call(t()) :: {:ok, t()} | {:error, any()}
+end

lib/k8s/middleware/request/base_url.ex

@@ -0,0 +1,18 @@
+# defmodule K8s.Middleware.Request.BaseURL do
+#   @behaviour K8s.Middleware.Request
+
+#   @doc """
+
+#   ## Examples
+#       iex> conn = K8s.Conn.from_file("./test/support/kube-config.yaml")
+#       ...> K8s.Cluster.Registry.add(:test_cluster, conn)
+#       ...> request = %K8s.Middleware.Request{cluster: :test_cluster}
+#       ...> K8s.Middleware.Request.BaseURL.call(request)      
+#       {:ok, %K8s.Middleware.Request{cluster: :test_cluster, url: "https://localhost:6443"}}
+#   """
+#   @impl true
+#   def call(%K8s.Middleware.Request{} = req) do
+#     {:ok, url} <- Cluster.url_for(operation, cluster_name)
+#     {:ok, req}
+#   end
+# end

lib/k8s/middleware/request/encode_body.ex

@@ -0,0 +1,25 @@
+defmodule K8s.Middleware.Request.EncodeBody do
+  @moduledoc """
+  Naive JSON body encoder.
+
+  Encodes JSON payloads when given an modifiying HTTP verb, otherwise returns an empty string.
+  """
+  @behaviour K8s.Middleware.Request
+  alias K8s.Middleware.Request
+
+  @impl true
+  def call(%Request{method: method, body: body} = req) do
+    case encode(body, method) do
+      {:ok, encoded_body} ->
+        req = %Request{req | body: encoded_body}
+        {:ok, req}
+
+      error ->
+        error
+    end
+  end
+
+  @spec encode(any(), atom()) :: {:ok, binary} | {:error, any}
+  defp encode(body, http_method) when http_method in [:put, :patch, :post], do: Jason.encode(body)
+  defp encode(_, _), do: {:ok, ""}
+end

lib/k8s/middleware/request/initialize.ex

@@ -0,0 +1,19 @@
+defmodule K8s.Middleware.Request.Initialize do
+  @moduledoc """
+  Initializes a request with connection details (header and HTTPoison opts) from `K8s.Conn.RequestOptions`
+  """
+  @behaviour K8s.Middleware.Request
+  alias K8s.Middleware.Request
+
+  @impl true
+  def call(%Request{cluster: cluster, method: method, headers: headers, opts: opts} = req) do
+    with {:ok, conn} <- K8s.Cluster.conn(cluster),
+         {:ok, request_options} <- K8s.Conn.RequestOptions.generate(conn) do
+      request_option_headers = K8s.http_provider().headers(method, request_options)
+      updated_headers = Keyword.merge(headers, request_option_headers)
+      updated_opts = Keyword.merge([ssl: request_options.ssl_options], opts)
+      updated_request = %Request{req | headers: updated_headers, opts: updated_opts}
+      {:ok, updated_request}
+    end
+  end
+end