807e93631268e5fd52ca29e3e4755088cf11bf27-PR-262

pending completion

Build # 807e93631268e5fd52ca29e3e4755088cf11bf27-PR-262

Build Type

Pull #262

github

Committed by

mruoss

Commit Message

add possibility to wait for delete

Pull Request Pull Request #262: add possibility to wait for delete

Run Details

6 of 6 new or added lines in 1 file covered. (100.0%)

732 of 1009 relevant lines covered (72.55%)

44.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

59.02

/lib/k8s/client/runner/base.ex

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

1	defmodule K8s.Client.Runner.Base do
2	@moduledoc """
3	Base HTTP processor for `K8s.Client`.
4	"""
5
6	@type error_t ::
7	{:error, K8s.Middleware.Error.t()}
8	\| {:error, K8s.Operation.Error.t()}
9	\| {:error, K8s.Client.APIError.t()}
10	\| {:error, K8s.Discovery.Error.t()}
11	\| {:error, atom()}
12	\| {:error, binary()}
13	@type result_t :: {:ok, map() \| reference()} \| error_t
14
15	@typedoc "Acceptable HTTP body types"
16	@type body_t :: list(map()) \| map() \| binary() \| nil
17
18	alias K8s.Conn
19	alias K8s.Middleware.Request
20	alias K8s.Operation
21
22	require Logger
23
24	@doc """
25	Runs a `K8s.Operation`.
26
27	## Examples
28
29	Note: Examples assume a `K8s.Conn` was configured named `"test"`. See `K8s.Conn.Config`.
30
31	Running a list pods operation:
32
33	```elixir
34	{:ok, conn} = K8s.Conn.from_file("test/support/kube-config.yaml")
35	operation = K8s.Client.list("v1", "Pod", namespace: :all)
36	{:ok, %{"items" => pods}} = K8s.Client.run(conn, operation)
37	```
38
39	Running a dry-run of a create deployment operation:
40
41	```elixir
42	deployment = %{
43	"apiVersion" => "apps/v1",
44	"kind" => "Deployment",
45	"metadata" => %{
46	"labels" => %{
47	"app" => "nginx"
48	},
49	"name" => "nginx",
50	"namespace" => "test"
51	},
52	"spec" => %{
53	"replicas" => 2,
54	"selector" => %{
55	"matchLabels" => %{
56	"app" => "nginx"
57	}
58	},
59	"template" => %{
60	"metadata" => %{
61	"labels" => %{
62	"app" => "nginx"
63	}
64	},
65	"spec" => %{
66	"containers" => %{
67	"image" => "nginx",
68	"name" => "nginx"
69	}
70	}
71	}
72	}
73	}
74
75	operation =
76	deployment
77	\|> K8s.Client.create()
78	\|> K8s.Operation.put_query_param(:dryRun, "all")
79
80	{:ok, conn} = K8s.Conn.from_file("test/support/kube-config.yaml")
81	{:ok, result} = K8s.Client.Runner.Base.run(conn, operation)
82	```
83	"""
84	@spec run(Operation.t()) :: result_t
85	def run(%Operation{conn: %Conn{} = conn} = operation),
86	do: run(conn, operation, [])	5✔
87
88	@spec run(Conn.t(), Operation.t()) :: result_t
89	def run(%Conn{} = conn, %Operation{} = operation),
90	do: run(conn, operation, [])	69✔
91
92	@spec run(Operation.t(), keyword()) :: result_t
93	def run(%Operation{conn: %Conn{} = conn} = operation, http_opts),
94	do: run(conn, operation, http_opts)	×
95
96	@doc """
97	Run a connect operation and pass `websocket_driver_opts` to `K8s.Client.WebSocketProvider`
98	See `run/3`
99	"""
100	@spec run(Conn.t(), Operation.t(), keyword()) :: result_t
101	def run(_conn, %Operation{verb: :watch}, _) do
102	msg = "Watch operations have to be streamed. Use K8s.Client.stream/N"	×
103
104	{:error, %K8s.Operation.Error{message: msg}}
105	end
106
107	# Run an operation and pass `http_opts` to `K8s.Client.HTTPProvider`
108	def run(%Conn{} = conn, %Operation{verb: :connect} = operation, http_opts) do
109	with {:ok, url} <- K8s.Discovery.url_for(conn, operation),	×
110	req <- new_request(conn, url, operation, operation.data, http_opts),	×
111	{:ok, req} <- K8s.Middleware.run(req, conn.middleware.request) do	×
112	conn.http_provider.websocket_request(	×
113	req.uri,	×
114	Keyword.merge(req.headers, Accept: "/"),	×
115	req.opts	×
116	)
117	end
118	end
119
120	# Run an operation and pass `http_opts` to `K8s.Client.HTTPProvider`
121	def run(%Conn{} = conn, %Operation{} = operation, http_opts) do
122	with {:ok, url} <- K8s.Discovery.url_for(conn, operation),	77✔
123	req <- new_request(conn, url, operation, operation.data, http_opts),	77✔
124	{:ok, req} <- K8s.Middleware.run(req, conn.middleware.request) do	77✔
125	conn.http_provider.request(req.method, req.uri, req.body, req.headers, req.opts)	77✔
126	end
127	end
128
129	@doc """
130	Runs a `K8s.Operation` and streams the response.
131	"""
132	@spec stream(Conn.t(), Operation.t(), keyword()) :: K8s.Client.Provider.stream_response_t()
133	def stream(%Conn{} = conn, %Operation{verb: :connect} = operation, http_opts) do
134	with {:ok, url} <- K8s.Discovery.url_for(conn, operation),	2✔
135	req <- new_request(conn, url, operation, operation.data, http_opts),	2✔
136	{:ok, req} <- K8s.Middleware.run(req, conn.middleware.request) do	2✔
137	conn.http_provider.websocket_stream(	2✔
138	req.uri,	2✔
139	Keyword.merge(req.headers, Accept: "/"),	2✔
140	req.opts	2✔
141	)
142	end
143	end
144
145	def stream(%Conn{} = conn, %Operation{} = operation, http_opts) do
146	with {:ok, url} <- K8s.Discovery.url_for(conn, operation),	15✔
147	req <- new_request(conn, url, operation, operation.data, http_opts),	15✔
148	{:ok, req} <- K8s.Middleware.run(req, conn.middleware.request) do	15✔
149	conn.http_provider.stream(	15✔
150	req.method,	15✔
151	req.uri,	15✔
152	req.body,	15✔
153	req.headers,	15✔
154	req.opts	15✔
155	)
156	end
157	end
158
159	@doc """
160	Runs a `K8s.Operation` and streams the response.
161	"""
162	@spec stream_to(Conn.t(), Operation.t(), keyword(), pid()) ::
163	K8s.Client.Provider.stream_to_response_t()
164	def stream_to(%Conn{} = conn, %Operation{verb: :connect} = operation, http_opts, stream_to) do
165	with {:ok, url} <- K8s.Discovery.url_for(conn, operation),	×
166	req <- new_request(conn, url, operation, operation.data, http_opts),	×
167	{:ok, req} <- K8s.Middleware.run(req, conn.middleware.request) do	×
168	conn.http_provider.websocket_stream_to(	×
169	req.uri,	×
170	Keyword.merge(req.headers, Accept: "/"),	×
171	req.opts,	×
172	stream_to
173	)
174	end
175	end
176
177	def stream_to(%Conn{} = conn, %Operation{} = operation, http_opts, stream_to) do
178	with {:ok, url} <- K8s.Discovery.url_for(conn, operation),	×
179	req <- new_request(conn, url, operation, operation.data, http_opts),	×
180	{:ok, req} <- K8s.Middleware.run(req, conn.middleware.request) do	×
181	conn.http_provider.stream_to(	×
182	req.method,	×
183	req.uri,	×
184	req.body,	×
185	req.headers,	×
186	req.opts,	×
187	stream_to
188	)
189	end
190	end
191
192	@spec new_request(
193	Conn.t(),
194	binary(),
195	Operation.t(),
196	String.t() \| nil,
197	keyword()
198	) :: Request.t()
199	defp new_request(%Conn{} = conn, url, %Operation{} = operation, body, http_opts) do
200	req = %Request{conn: conn, method: operation.method, body: body}	94✔
201
202	headers = operation.header_params	94✔
203	operation_query_params = build_query_params(operation)	94✔
204	http_opts_params = Keyword.get(http_opts, :params, [])	94✔
205	merged_params = Keyword.merge(operation_query_params, http_opts_params)	94✔
206
207	uri = url \|> URI.parse() \|> URI.append_query(URI.encode_query(merged_params))	94✔
208
209	%Request{req \| opts: http_opts, headers: headers, uri: uri}	94✔
210	end
211
212	@spec build_query_params(Operation.t()) :: String.t() \| keyword()
213	defp build_query_params(operation) do
214	{commands, query_params} = Keyword.pop_values(operation.query_params, :command)	94✔
215
216	commands =	94✔
217	commands
218	\|> List.flatten()
219	\|> Enum.map(&{:command, &1})	4✔
220
221	selector = Operation.get_selector(operation)	94✔
222
223	selectors =	94✔
224	[
225	labelSelector: K8s.Selector.labels_to_s(selector),
226	fieldSelector: K8s.Selector.fields_to_s(selector)
227	]
228	\|> Keyword.reject(&(elem(&1, 1) == ""))	188✔
229
230	query_params
231	\|> Keyword.delete(:labelSelector)
232	\|> Keyword.merge(selectors)
233	\|> Keyword.merge(commands)	94✔
234	end
235	end

coryodaniel / k8s / 807e93631268e5fd52ca29e3e4755088cf11bf27-PR-262

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous