refactor k8s list nodes into Node module

README.md

@@ -168,8 +168,8 @@ kind: EvictionPolicy
 metadata:
   name: unpreferred-nodes-nginx
 spec:
-  mode: unpreferred 
-  maxLifetime: 600 
+  mode: unpreferred
+  maxLifetime: 600
   selector:
     matchLabels:
       app: nginx

lib/ballast/controllers/v1/eviction_policy.ex

@@ -47,16 +47,13 @@ defmodule Ballast.Controller.V1.EvictionPolicy do
   @impl Bonny.Controller
   def reconcile(payload) do
     handle_eviction(payload)
-    :ok
   end
 
-  @spec handle_eviction(map()) :: :ok
+  @spec handle_eviction(map()) :: :ok | :error
   defp handle_eviction(%{} = policy) do
     with {:ok, pods} <- Ballast.Evictor.evictable(policy) do
       Enum.each(pods, &Ballast.Kube.Eviction.create/1)
     end
-
-    :ok
   end
 
   defp handle_eviction(_) do

lib/ballast/evictor.ex

@@ -3,6 +3,7 @@ defmodule Ballast.Evictor do
   Finds pods that are candidates for eviction.
   """
 
+  @node_batch_size 100
   @pod_batch_size 500
   @default_max_lifetime 600
 
@@ -43,7 +44,7 @@ defmodule Ballast.Evictor do
   """
   @spec evictable(map) :: {:ok, list(map)} | {:error, HTTPoison.Response.t()}
   def evictable(%{} = policy) do
-    with {:ok, nodes} <- get_nodes(),
+    with {:ok, nodes} <- Ballast.Kube.Node.list(%{limit: @node_batch_size}),
          {:ok, candidates} <- candidates(policy) do
       max_lifetime = max_lifetime(policy)
       started_before = pods_started_before(candidates, max_lifetime)
@@ -136,13 +137,4 @@ defmodule Ballast.Evictor do
   defp parse_seconds(sec) when is_integer(sec), do: sec
   defp parse_seconds({sec, _}), do: sec
   defp parse_seconds(_), do: 0
-
-  defp get_nodes() do
-    op = K8s.Client.list("v1", :nodes)
-
-    with {:ok, stream} <- K8s.Client.stream(op, :default) do
-      nodes = Enum.into(stream, [])
-      {:ok, nodes}
-    end
-  end
 end

lib/ballast/kube/node.ex

@@ -13,7 +13,28 @@ defmodule Ballast.Kube.Node do
     "DiskPressure"
   ]
 
-  alias K8s.Resource
+  alias K8s.{Client, Resource}
+  alias Ballast.Sys.Instrumentation, as: Inst
+
+  @doc """
+  List kubernetes nodes.
+  """
+  @spec list(map()) :: {:ok, list(map)} | :error
+  def list(params \\ %{}) do
+    op = Client.list("v1", :nodes)
+
+    with {:ok, stream} <- Client.stream(op, :default, params: params) do
+      {duration, nodes} = :timer.tc(Enum, :into, [stream, []])
+      measurements = %{duration: duration, count: length(nodes)}
+      Inst.nodes_list_succeeded(measurements, %{})
+
+      {:ok, nodes}
+    else
+      _error ->
+        Inst.nodes_list_failed(%{}, %{})
+        :error
+    end
+  end
 
   @doc """
   Checks if `status.conditions` are present and node is `Ready`
@@ -77,20 +98,20 @@ defmodule Ballast.Kube.Node do
 
   ## Examples
     Matching all expressions:
-    
+
       iex> node = %{"kind" => "Node", "metadata" => %{"labels" => %{"env" => "prod", "tier" => "frontend"}}}
       ...> expr1 = %{"operator" => "In", "key" => "env", "values" => ["prod", "qa"]}
       ...> expr2 = %{"operator" => "Exists", "key" => "tier"}
       ...> Ballast.Kube.Node.match_expressions?(node, [expr1, expr2])
       true
-      
+
     Matching some expressions:
-    
+
       iex> node = %{"kind" => "Node", "metadata" => %{"labels" => %{"env" => "prod", "tier" => "frontend"}}}
       ...> expr1 = %{"operator" => "In", "key" => "env", "values" => ["prod", "qa"]}
       ...> expr2 = %{"operator" => "Exists", "key" => "foo"}
       ...> Ballast.Kube.Node.match_expressions?(node, [expr1, expr2])
-      false      
+      false
   """
   @spec match_expressions?(map, list(map)) :: boolean
   def match_expressions?(node, exprs) do
@@ -106,48 +127,48 @@ defmodule Ballast.Kube.Node do
       ...> expr = %{"operator" => "In", "key" => "env", "values" => ["prod", "qa"]}
       ...> Ballast.Kube.Node.match_expression?(node, expr)
       true
-    
+
     When an `In` expression doesnt match
       iex> node = %{"kind" => "Node", "metadata" => %{"labels" => %{"env" => "dev"}}}
       ...> expr = %{"operator" => "In", "key" => "env", "values" => ["prod", "qa"]}
       ...> Ballast.Kube.Node.match_expression?(node, expr)
       false
-      
+
     When an `NotIn` expression matches
       iex> node = %{"kind" => "Node", "metadata" => %{"labels" => %{"env" => "dev"}}}
       ...> expr = %{"operator" => "NotIn", "key" => "env", "values" => ["prod"]}
       ...> Ballast.Kube.Node.match_expression?(node, expr)
       true
-    
+
     When an `NotIn` expression doesnt match
       iex> node = %{"kind" => "Node", "metadata" => %{"labels" => %{"env" => "dev"}}}
       ...> expr = %{"operator" => "NotIn", "key" => "env", "values" => ["dev"]}
       ...> Ballast.Kube.Node.match_expression?(node, expr)
       false
-      
+
     When an `Exists` expression matches
       iex> node = %{"kind" => "Node", "metadata" => %{"labels" => %{"env" => "dev"}}}
       ...> expr = %{"operator" => "Exists", "key" => "env"}
       ...> Ballast.Kube.Node.match_expression?(node, expr)
       true
-      
+
     When an `Exists` expression doesnt match
       iex> node = %{"kind" => "Node", "metadata" => %{"labels" => %{"env" => "dev"}}}
       ...> expr = %{"operator" => "Exists", "key" => "tier"}
       ...> Ballast.Kube.Node.match_expression?(node, expr)
       false
-      
+
     When an `DoesNotExist` expression matches
       iex> node = %{"kind" => "Node", "metadata" => %{"labels" => %{"env" => "dev"}}}
       ...> expr = %{"operator" => "DoesNotExist", "key" => "tier"}
       ...> Ballast.Kube.Node.match_expression?(node, expr)
       true
-      
+
     When an `DoesNotExist` expression doesnt match
       iex> node = %{"kind" => "Node", "metadata" => %{"labels" => %{"env" => "dev"}}}
       ...> expr = %{"operator" => "DoesNotExist", "key" => "env"}
       ...> Ballast.Kube.Node.match_expression?(node, expr)
-      false      
+      false
   """
   @spec match_expression?(map(), map()) :: boolean()
   def match_expression?(node, %{"operator" => "In", "key" => k, "values" => v}) do

lib/ballast/node_pool.ex

@@ -150,28 +150,27 @@ defmodule Ballast.NodePool do
   Determine if a pool is under pressure.
 
   A pool is considered under pressure when more than #{@node_pool_pressure_percent} percent of its nodes are under pressure.
+
+  Notes:
+  * Assumes there is pressure if there is an error from the k8s API.
+  * Considers no nodes returned as under pressure.
   """
   @spec under_pressure?(Ballast.NodePool.t()) :: boolean()
   def under_pressure?(%Ballast.NodePool{} = pool) do
-    {:ok, stream} = nodes(pool)
-    nodes = Enum.into(stream, [])
-
-    nodes_under_pressure = Enum.filter(nodes, fn node -> node_under_pressure?(node) end)
+    label_selector = adapter_for(pool).label_selector(pool)
+    params = %{labelSelector: label_selector}
 
-    percent_under_pressure = length(nodes_under_pressure) / length(nodes)
-    percent_under_pressure >= @node_pool_pressure_threshold
-  end
+    with {:ok, [_h | _t] = nodes} <- Ballast.Kube.Node.list(params) do
+      nodes_under_pressure = Enum.filter(nodes, fn node -> node_under_pressure?(node) end)
 
-  @doc """
-  Get the nodes from the kubernetes API matching the provider's label selector.
-  """
-  @spec nodes(Ballast.NodePool.t()) :: {:ok, Enumerable.t()} | {:error, atom()}
-  def nodes(%Ballast.NodePool{} = pool) do
-    label_selector = adapter_for(pool).label_selector(pool)
-    op = K8s.Client.list("v1", :nodes)
-    K8s.Client.stream(op, :default, params: %{labelSelector: label_selector})
+      percent_under_pressure = length(nodes_under_pressure) / length(nodes)
+      percent_under_pressure >= @node_pool_pressure_threshold
+    else
+      _error -> true
+    end
   end
 
+  @spec node_under_pressure?(map) :: boolean()
   defp node_under_pressure?(node) do
     !Ballast.Kube.Node.ready?(node) || Ballast.Kube.Node.resources_constrained?(node)
   end
@@ -180,6 +179,7 @@ defmodule Ballast.NodePool do
   @spec measurements_and_metadata(Changeset.t()) :: {map, map}
   defp measurements_and_metadata(changeset) do
     measurements = %{
+      source_pool_current_count: changeset.source_count,
       managed_pool_current_count: changeset.pool.instance_count,
       managed_pool_current_minimum_count: changeset.pool.minimum_count,
       managed_pool_new_minimum_count: changeset.minimum_count

lib/ballast/pool_policy/changeset.ex

@@ -6,11 +6,18 @@ defmodule Ballast.PoolPolicy.Changeset do
   alias Ballast.NodePool
   alias Ballast.PoolPolicy.{Changeset, ManagedPool}
 
-  defstruct [:pool, :minimum_count, :strategy]
+  defstruct [:pool, :minimum_count, :source_count, :strategy]
 
+  @typedoc """
+  * `pool` - the managed pool changeset will be applied to
+  * `minimum_count` - the new minimum count for the autoscaler or cluster
+  * `source_count` - the current count of the source pool
+  * `strategy` - what ballast thinks is happening to the source pool - poor name.
+  """
   @type t :: %__MODULE__{
           pool: NodePool.t(),
-          minimum_count: pos_integer,
+          source_count: integer,
+          minimum_count: integer,
           strategy: :nothing | :scale_up | :scale_down
         }
 
@@ -21,7 +28,7 @@ defmodule Ballast.PoolPolicy.Changeset do
       iex> managed_pool = %Ballast.PoolPolicy.ManagedPool{pool: %Ballast.NodePool{name: "managed-pool"}, minimum_percent: 30, minimum_instances: 1}
       ...> source_pool = %Ballast.NodePool{instance_count: 10}
       ...> Ballast.PoolPolicy.Changeset.new(managed_pool, source_pool)
-      %Ballast.PoolPolicy.Changeset{minimum_count: 3, pool: %Ballast.NodePool{cluster: nil, data: nil, instance_count: nil, location: nil, name: "managed-pool", project: nil, under_pressure: nil}, strategy: :scale_down}
+      %Ballast.PoolPolicy.Changeset{source_count: 10, minimum_count: 3, pool: %Ballast.NodePool{cluster: nil, data: nil, instance_count: nil, location: nil, name: "managed-pool", project: nil, under_pressure: nil}, strategy: :scale_down}
   """
   @spec new(ManagedPool.t(), NodePool.t()) :: t
   def new(managed_pool, %NodePool{instance_count: source_count} = source_pool) do
@@ -30,6 +37,7 @@ defmodule Ballast.PoolPolicy.Changeset do
 
     %Changeset{
       pool: managed_pool.pool,
+      source_count: source_count,
       minimum_count: calculated_minimum_count,
       strategy: strategy(managed_pool.pool, source_pool)
     }

lib/ballast/sys/instrumentation.ex

@@ -2,6 +2,12 @@ defmodule Ballast.Sys.Instrumentation do
   @moduledoc false
   use Notion, name: :ballast, metadata: %{}
 
+  @doc "Get nodes succceeded"
+  defevent([:nodes, :list, :succeeded])
+
+  @doc "Get nodes failed"
+  defevent([:nodes, :list, :failed])
+
   @doc "Pod eviction succceeded"
   defevent([:pod, :eviction, :succeeded])
 

lib/ballast/sys/metrics.ex

@@ -15,6 +15,8 @@ defmodule Ballast.Sys.Metrics do
       counter("ballast.pool_policy.deleted.count", description: nil),
       counter("ballast.pool_policy.modified.count", description: nil),
       counter("ballast.pool_policy.added.count", description: nil),
+      counter("ballast.nodes.list.succeeded.count", description: nil),
+      counter("ballast.nodes.list.failed.count", description: nil),
       counter("ballast.pod.eviction.failed.count", description: nil),
       counter("ballast.pod.eviction.succeeded.count", description: nil),
       counter("ballast.get_eviction_candidates.failed.count", description: nil),