Fix Kubernetes strategy crash on endpoint addresses without targetRef

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## Unreleased
 
 - Add `kubernetes_use_cached_resources` option to Kubernetes strategy
+- Fix Kubernetes strategy crash when an endpoints address has no `targetRef`
 
 ## 3.4.1
 

lib/strategy/kubernetes.ex

@@ -409,7 +409,7 @@ defmodule Cluster.Strategy.Kubernetes do
 
         case :httpc.request(:get, {~c"https://#{master}/#{path}", headers}, http_options, []) do
           {:ok, {{_version, 200, _status}, _headers, body}} ->
-            parse_response(ip_lookup_mode, Jason.decode!(body))
+            parse_response(ip_lookup_mode, Jason.decode!(body), namespace)
             |> Enum.map(fn node_info ->
               format_node(
                 Keyword.get(config, :mode, :ip),
@@ -462,17 +462,22 @@ defmodule Cluster.Strategy.Kubernetes do
 
   defp apply_param(params, _key, _value), do: params
 
-  defp parse_response(:endpoints, resp) do
+  defp parse_response(:endpoints, resp, namespace) do
     case resp do
       %{"items" => items} when is_list(items) ->
         Enum.reduce(items, [], fn
           %{"subsets" => subsets}, acc when is_list(subsets) ->
             addrs =
               Enum.flat_map(subsets, fn
                 %{"addresses" => addresses} when is_list(addresses) ->
-                  Enum.map(addresses, fn %{"ip" => ip, "targetRef" => %{"namespace" => namespace}} =
-                                           address ->
-                    %{ip: ip, namespace: namespace, hostname: address["hostname"]}
+                  Enum.map(addresses, fn %{"ip" => ip} = address ->
+                    # targetRef is optional, and endpoints which are not backed by a pod
+                    # (manually managed ones, for example) do not have one
+                    %{
+                      ip: ip,
+                      namespace: get_in(address, ["targetRef", "namespace"]) || namespace,
+                      hostname: address["hostname"]
+                    }
                   end)
 
                 _ ->
@@ -490,7 +495,7 @@ defmodule Cluster.Strategy.Kubernetes do
     end
   end
 
-  defp parse_response(:pods, resp) do
+  defp parse_response(:pods, resp, _namespace) do
     case resp do
       %{"items" => items} when is_list(items) ->
         Enum.map(items, fn

test/fixtures/vcr_cassettes/kubernetes_no_target_ref.json

@@ -0,0 +1,34 @@
+[
+  {
+    "request": {
+      "body": "",
+      "headers": {
+        "authorization": "***"
+      },
+      "method": "get",
+      "options": {
+        "httpc_options": [],
+        "http_options": {
+          "ssl": "[verify: :verify_none]"
+        }
+      },
+      "request_body": "",
+      "url": "https://cluster.localhost./api/v1/namespaces/__libcluster_test/endpoints?labelSelector=app=test_selector"
+    },
+    "response": {
+      "binary": false,
+      "body": "{\"kind\": \"EndpointsList\", \"apiVersion\": \"v1\", \"metadata\": {\"selfLink\": \"SELFLINK_PLACEHOLDER\", \"resourceVersion\": \"17042410\"}, \"items\": [{\"metadata\": {\"name\": \"development-development\", \"namespace\": \"__libcluster_test\", \"selfLink\": \"SELFLINK_PLACEHOLDER\", \"uid\": \"7e3faf1e-0294-11e8-bcad-42010a9c01cc\", \"resourceVersion\": \"17037787\", \"creationTimestamp\": \"2018-01-26T12:29:03Z\", \"labels\": {\"app\": \"test_selector\", \"chart\": \"CHART_PLACEHOLDER\"}}, \"subsets\": [{\"addresses\": [{\"hostname\": \"my-hostname-0\", \"ip\": \"10.48.33.136\", \"nodeName\": \"gke-jshmrtn-cluster-default-pool-a61da41f-db9x\", \"targetRef\": {\"kind\": \"Pod\", \"namespace\": \"airatel-service-localization\", \"name\": \"development-4292695165-mgq9f\", \"uid\": \"eb0f3e80-0295-11e8-bcad-42010a9c01cc\", \"resourceVersion\": \"17037783\"}}, {\"ip\": \"10.48.33.137\", \"nodeName\": \"gke-jshmrtn-cluster-default-pool-a61da41f-db9x\"}], \"ports\": [{\"name\": \"web\", \"port\": 8443, \"protocol\": \"TCP\"}]}]}]}",
+      "headers": {
+        "date": "Fri, 26 Jan 2018 13:18:46 GMT",
+        "content-length": "932",
+        "content-type": "application/json"
+      },
+      "status_code": [
+        "HTTP/1.1",
+        200,
+        "OK"
+      ],
+      "type": "ok"
+    }
+  }
+]

test/kubernetes_test.exs

@@ -283,6 +283,41 @@ defmodule Cluster.Strategy.KubernetesTest do
       end
     end
 
+    test "works with endpoints whose addresses have no targetRef" do
+      use_cassette "kubernetes_no_target_ref", custom: true do
+        capture_log(fn ->
+          start_supervised!({Kubernetes,
+           [
+             %Cluster.Strategy.State{
+               topology: :name,
+               config: [
+                 kubernetes_node_basename: "test_basename",
+                 kubernetes_selector: "app=test_selector",
+                 mode: :dns,
+                 # If you want to run the test freshly, you'll need to create a DNS Entry
+                 kubernetes_master: "cluster.localhost.",
+                 kubernetes_service_account_path:
+                   Path.join([__DIR__, "fixtures", "kubernetes", "service_account"])
+               ],
+               connect: {Nodes, :connect, [self()]},
+               disconnect: {Nodes, :disconnect, [self()]},
+               list_nodes: {Nodes, :list_nodes, [[]]}
+             }
+           ]})
+
+          # address with a targetRef keeps the namespace it points at
+          assert_receive {:connect,
+                          :"test_basename@10-48-33-136.airatel-service-localization.pod.cluster.local"},
+                         5_000
+
+          # address without a targetRef falls back to the queried namespace
+          assert_receive {:connect,
+                          :"test_basename@10-48-33-137.__libcluster_test.pod.cluster.local"},
+                         5_000
+        end)
+      end
+    end
+
     test "works with pods and dns" do
       use_cassette "kubernetes_pods", custom: true do
         capture_log(fn ->