Merge pull request #255 from coryodaniel/fix-connection-op-for-logs

Fix connection op for pods/log

CHANGELOG.md

@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- `K8s.Client.connect/4` - Support connecting to `pods/log` subresource. - [#254](https://github.com/coryodaniel/k8s/issues/254), [#255](https://github.com/coryodaniel/k8s/issues/255)
 - `K8s.Conn.from_env/2` - Generates configuration from a file defined by an env variable. - [#251](https://github.com/coryodaniel/k8s/pull/251)
 - `K8s.Conn` - Better hexdocs
 

guides/operations.md

@@ -177,11 +177,12 @@ conn
 |> Enum.into([])
 ```
 
-## Connect to pods and execute commands
+## Connect to `pods/exec` subresource and execute commands
 
-The `:connect` operation is used to connect to pods and execute commands.
-A `:connect` operation is created with `K8s.Client.connect/N`. Be sure to pass
-the command you want to run in the options.
+Use the `:connect` operation to connect to the `pods/exec` subresource and
+execute commands. A `:connect` operation is created with `K8s.Client.connect/N`.
+When connecting to `pods/exec`, be sure to pass the command you want to run in
+the options.
 
 ### Waiting for command termination
 
@@ -263,3 +264,39 @@ commands). See the example below.
 
   {:ok, response} = K8s.Client.run(conn, op)
 ```
+
+## Connect to `pods/log` subresource to read logs from Pods
+
+Use the `:connect` operation to connect to the `pods/log` subresource. A
+`:connect` operation is created with `K8s.Client.connect/N`.
+
+### Options
+
+Refer to the [Kubernetes
+documentation](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.27/#read-log-pod-v1-core)
+for documentation on these options.
+
+- `container`
+- `follow` - Use with `K8s.Client.stream/N` or `K8s.Client.stream_to/N`.
+- `insecureSkipTLSVerifyBackend`
+- `limitBytes`
+- `pretty`
+- `previous`
+- `sinceSeconds`
+- `tailLines`
+- `timestamps`
+
+```elixir
+  {:ok, conn} = K8s.Conn.from_file("~/.kube/config")
+
+  {:ok, stream} = K8s.Client.connect(
+    "v1",
+    "pods/log",
+    [namespace: "default", name: "nginx-8f458dc5b-zwmkb"],
+    command: ["/bin/sh", "-c", "nginx -t"],
+    container: "main",
+    follow: true
+  )
+  |> K8s.Client.put_conn(conn)
+  |> K8s.Client.stream()
+```

lib/k8s/client/mint/request.ex

@@ -132,6 +132,7 @@ defmodule K8s.Client.Mint.Request do
   def map_frame({:binary, <<1, msg::binary>>}), do: {:stdout, msg}
   def map_frame({:binary, <<2, msg::binary>>}), do: {:stderr, msg}
   def map_frame({:binary, <<3, msg::binary>>}), do: {:error, msg}
+  def map_frame({:binary, msg}), do: {:stdout, msg}
 
   @spec map_outgoing_frame({:stdin, binary()} | {:close, integer(), binary()} | :close | :exit) ::
           {:ok, :close | {:text, binary} | {:close, integer(), binary()}}

lib/k8s/operation.ex

@@ -30,6 +30,21 @@ defmodule K8s.Operation do
     apply: "application/apply-patch+yaml"
   }
 
+  @exec_default_params [stdin: true, stdout: true, stderr: true, tty: false]
+  @exec_allowed_connect_params [:stdin, :stdout, :stderr, :tty, :command, :container]
+
+  @log_allowed_connect_params [
+    :container,
+    :follow,
+    :insecureSkipTLSVerifyBackend,
+    :limitBytes,
+    :pretty,
+    :previous,
+    :sinceSeconds,
+    :tailLines,
+    :timestamps
+  ]
+
   defstruct method: nil,
             verb: nil,
             api_version: nil,
@@ -192,6 +207,7 @@ defmodule K8s.Operation do
     )
   end
 
+  # credo:disable-for-next-line Credo.Check.Refactor.CyclomaticComplexity
   def build(verb, api_version, name_or_kind, path_params, data, opts) do
     http_method = @verb_map[verb] || verb
     patch_type = Keyword.get(opts, :patch_type, :not_set)
@@ -204,17 +220,19 @@ defmodule K8s.Operation do
 
     query_params =
       cond do
-        verb === :patch && patch_type === :apply ->
+        verb === :patch and patch_type === :apply ->
           [
             fieldManager: Keyword.get(opts, :field_manager, "elixir"),
             force: Keyword.get(opts, :force, true)
           ]
 
-        verb === :connect ->
-          [stdin: true, stdout: true, stderr: true, tty: false]
-          |> Keyword.merge(
-            Keyword.take(opts, [:stdin, :stdout, :stderr, :tty, :command, :container])
-          )
+        verb === :connect and name_or_kind === "pods/exec" ->
+          @exec_default_params
+          |> Keyword.merge(opts)
+          |> Keyword.take(@exec_allowed_connect_params)
+
+        verb === :connect and name_or_kind === "pods/log" ->
+          Keyword.take(opts, @log_allowed_connect_params)
 
         true ->
           []

test/k8s/client/runner/base_integration_test.exs

@@ -257,7 +257,7 @@ defmodule K8s.Client.Runner.BaseIntegrationTest do
     end
 
     @tag :integration
-    test "creating a operation without correct kind `pod/exec` should return an error", %{
+    test "creating a operation without correct kind should return an error", %{
       conn: conn
     } do
       operation =
@@ -271,7 +271,7 @@ defmodule K8s.Client.Runner.BaseIntegrationTest do
 
   @tag :integration
   @tag :websocket
-  test "runs :connect operations and returns stdout", %{
+  test "runs :connect operations for pods/exec and returns stdout", %{
     conn: conn,
     labels: labels,
     test_id: test_id
@@ -303,6 +303,42 @@ defmodule K8s.Client.Runner.BaseIntegrationTest do
     assert response.stdout =~ "ok"
   end
 
+  @tag :integration
+  @tag :websocket
+  @tail_lines 5
+  test "runs :connect operations for pods/log and returns stdout", %{
+    conn: conn,
+    labels: labels,
+    test_id: test_id
+  } do
+    {:ok, created_pod} =
+      build_pod("k8s-ex-#{test_id}", labels)
+      |> K8s.Client.create()
+      |> K8s.Client.put_conn(conn)
+      |> K8s.Client.run()
+
+    {:ok, _} =
+      K8s.Client.wait_until(conn, K8s.Client.get(created_pod),
+        find: ["status", "containerStatuses", Access.filter(&(&1["ready"] == true))],
+        eval: &match?([_ | _], &1),
+        timeout: 60
+      )
+
+    {:ok, response} =
+      K8s.Client.connect(
+        created_pod["apiVersion"],
+        "pods/log",
+        [namespace: K8s.Resource.namespace(created_pod), name: K8s.Resource.name(created_pod)],
+        tailLines: @tail_lines
+      )
+      |> K8s.Client.put_conn(conn)
+      |> K8s.Client.run()
+
+    assert String.printable?(response.stdout)
+    lines = String.split(response.stdout, "\n", trim: true)
+    assert @tail_lines == length(lines)
+  end
+
   @tag :integration
   @tag :websocket
   test "runs :connect operations and returns errors", %{