defmodule K8s.Client.Runner.Base do
@moduledoc """
Base HTTP processor for `K8s.Client`.
"""
@type error_t ::
{:error, K8s.Middleware.Error.t()}
| {:error, K8s.Operation.Error.t()}
| {:error, K8s.Client.APIError.t()}
| {:error, K8s.Discovery.Error.t()}
| {:error, atom()}
| {:error, binary()}
@type result_t :: {:ok, map() | reference()} | error_t
@typedoc "Acceptable HTTP body types"
@type body_t :: list(map()) | map() | binary() | nil
alias K8s.Conn
alias K8s.Middleware.Request
alias K8s.Operation
require Logger
@doc """
Runs a `K8s.Operation`.
## Examples
*Note:* Examples assume a `K8s.Conn` was configured named `"test"`. See `K8s.Conn.Config`.
Running a list pods operation:
```elixir
{:ok, conn} = K8s.Conn.from_file("test/support/kube-config.yaml")
operation = K8s.Client.list("v1", "Pod", namespace: :all)
{:ok, %{"items" => pods}} = K8s.Client.run(conn, operation)
```
Running a dry-run of a create deployment operation:
```elixir
deployment = %{
"apiVersion" => "apps/v1",
"kind" => "Deployment",
"metadata" => %{
"labels" => %{
"app" => "nginx"
},
"name" => "nginx",
"namespace" => "test"
},
"spec" => %{
"replicas" => 2,
"selector" => %{
"matchLabels" => %{
"app" => "nginx"
}
},
"template" => %{
"metadata" => %{
"labels" => %{
"app" => "nginx"
}
},
"spec" => %{
"containers" => %{
"image" => "nginx",
"name" => "nginx"
}
}
}
}
}
operation =
deployment
|> K8s.Client.create()
|> K8s.Operation.put_query_param(:dryRun, "all")
{:ok, conn} = K8s.Conn.from_file("test/support/kube-config.yaml")
{:ok, result} = K8s.Client.Runner.Base.run(conn, operation)
```
"""
@spec run(Operation.t()) :: result_t
def run(%Operation{conn: %Conn{} = conn} = operation),
do: run(conn, operation, [])
@spec run(Conn.t(), Operation.t()) :: result_t
def run(%Conn{} = conn, %Operation{} = operation),
do: run(conn, operation, [])
@spec run(Operation.t(), keyword()) :: result_t
def run(%Operation{conn: %Conn{} = conn} = operation, http_opts),
do: run(conn, operation, http_opts)
@doc """
Run a connect operation and pass `websocket_driver_opts` to `K8s.Client.WebSocketProvider`
See `run/3`
"""
@spec run(Conn.t(), Operation.t(), keyword()) :: result_t
def run(_conn, %Operation{verb: :watch}, _) do
msg = "Watch operations have to be streamed. Use K8s.Client.stream/N"
{:error, %K8s.Operation.Error{message: msg}}
end
# Run an operation and pass `http_opts` to `K8s.Client.HTTPProvider`
def run(%Conn{} = conn, %Operation{verb: :connect} = operation, http_opts) do
with {:ok, url} <- K8s.Discovery.url_for(conn, operation),
req <- new_request(conn, url, operation, operation.data, http_opts),
{:ok, req} <- K8s.Middleware.run(req, conn.middleware.request) do
conn.http_provider.websocket_request(
req.uri,
Keyword.merge(req.headers, Accept: "*/*"),
req.opts
)
end
end
# Run an operation and pass `http_opts` to `K8s.Client.HTTPProvider`
def run(%Conn{} = conn, %Operation{} = operation, http_opts) do
with {:ok, url} <- K8s.Discovery.url_for(conn, operation),
req <- new_request(conn, url, operation, operation.data, http_opts),
{:ok, req} <- K8s.Middleware.run(req, conn.middleware.request) do
conn.http_provider.request(req.method, req.uri, req.body, req.headers, req.opts)
end
end
@doc """
Runs a `K8s.Operation` and streams the response.
"""
@spec stream(Conn.t(), Operation.t(), keyword()) :: K8s.Client.Provider.stream_response_t()
def stream(%Conn{} = conn, %Operation{verb: :connect} = operation, http_opts) do
with {:ok, url} <- K8s.Discovery.url_for(conn, operation),
req <- new_request(conn, url, operation, operation.data, http_opts),
{:ok, req} <- K8s.Middleware.run(req, conn.middleware.request) do
conn.http_provider.websocket_stream(
req.uri,
Keyword.merge(req.headers, Accept: "*/*"),
req.opts
)
end
end
def stream(%Conn{} = conn, %Operation{} = operation, http_opts) do
with {:ok, url} <- K8s.Discovery.url_for(conn, operation),
req <- new_request(conn, url, operation, operation.data, http_opts),
{:ok, req} <- K8s.Middleware.run(req, conn.middleware.request) do
conn.http_provider.stream(
req.method,
req.uri,
req.body,
req.headers,
req.opts
)
end
end
@doc """
Runs a `K8s.Operation` and streams the response.
"""
@spec stream_to(Conn.t(), Operation.t(), keyword(), pid()) ::
K8s.Client.Provider.stream_to_response_t()
def stream_to(%Conn{} = conn, %Operation{verb: :connect} = operation, http_opts, stream_to) do
with {:ok, url} <- K8s.Discovery.url_for(conn, operation),
req <- new_request(conn, url, operation, operation.data, http_opts),
{:ok, req} <- K8s.Middleware.run(req, conn.middleware.request) do
conn.http_provider.websocket_stream_to(
req.uri,
Keyword.merge(req.headers, Accept: "*/*"),
req.opts,
stream_to
)
end
end
def stream_to(%Conn{} = conn, %Operation{} = operation, http_opts, stream_to) do
with {:ok, url} <- K8s.Discovery.url_for(conn, operation),
req <- new_request(conn, url, operation, operation.data, http_opts),
{:ok, req} <- K8s.Middleware.run(req, conn.middleware.request) do
conn.http_provider.stream_to(
req.method,
req.uri,
req.body,
req.headers,
req.opts,
stream_to
)
end
end
@spec new_request(
Conn.t(),
binary(),
Operation.t(),
String.t() | nil,
keyword()
) :: Request.t()
defp new_request(%Conn{} = conn, url, %Operation{} = operation, body, http_opts) do
req = %Request{conn: conn, method: operation.method, body: body}
headers = operation.header_params
operation_query_params = build_query_params(operation)
http_opts_params = Keyword.get(http_opts, :params, [])
merged_params = Keyword.merge(operation_query_params, http_opts_params)
uri = url |> URI.parse() |> URI.append_query(URI.encode_query(merged_params))
%Request{req | opts: http_opts, headers: headers, uri: uri}
end
@spec build_query_params(Operation.t()) :: String.t() | keyword()
defp build_query_params(operation) do
{commands, query_params} = Keyword.pop_values(operation.query_params, :command)
commands =
commands
|> List.flatten()
|> Enum.map(&{:command, &1})
selector = Operation.get_selector(operation)
selectors =
[
labelSelector: K8s.Selector.labels_to_s(selector),
fieldSelector: K8s.Selector.fields_to_s(selector)
]
|> Keyword.reject(&(elem(&1, 1) == ""))
query_params
|> Keyword.delete(:labelSelector)
|> Keyword.merge(selectors)
|> Keyword.merge(commands)
end
end
