Add metrics filtering to devices query

lib/nerves_hub/devices.ex

@@ -17,6 +17,7 @@ defmodule NervesHub.Devices do
   alias NervesHub.Devices.Device
   alias NervesHub.Devices.DeviceCertificate
   alias NervesHub.Devices.DeviceHealth
+  alias NervesHub.Devices.DeviceMetric
   alias NervesHub.Devices.SharedSecretAuth
   alias NervesHub.Devices.InflightUpdate
   alias NervesHub.Devices.UpdatePayload
@@ -158,6 +159,39 @@ defmodule NervesHub.Devices do
 
   defp sort_devices(sort), do: sort
 
+  def filter_on_metric(query, key, value, operator) do
+    query
+    |> where(
+      [d],
+      d.id in subquery(metrics_query(key, value, operator))
+    )
+    |> preload_metric(key)
+  end
+
+  def metrics_query(key, value, operator) do
+    DeviceMetric
+    |> from
+    |> select([d], d.device_id)
+    |> where([dm], dm.key == ^key)
+    |> gt_or_lt(value, operator)
+    |> order_by(desc: :inserted_at)
+    |> distinct(:device_id)
+  end
+
+  defp gt_or_lt(query, value, :gt), do: where(query, [dm], dm.value > ^value)
+  defp gt_or_lt(query, value, :lt), do: where(query, [dm], dm.value < ^value)
+
+  defp preload_metric(query, key) do
+    preload_query =
+      DeviceMetric
+      |> distinct(:device_id)
+      |> where(key: ^key)
+      |> order_by([:device_id, desc: :inserted_at])
+
+    query
+    |> preload(device_metrics: ^preload_query)
+  end
+
   defp filtering(query, filters) do
     Enum.reduce(filters, query, fn {key, value}, query ->
       case {key, value} do
@@ -173,6 +207,12 @@ defmodule NervesHub.Devices do
         {:alarm, value} ->
           where(query, [d], d.id in subquery(Alarms.query_devices_with_alarm(value)))
 
+        {:metrics, %{key: key, value: value, operator: operator}} ->
+          filter_on_metric(query, key, value, operator)
+
+        {:metrics, _} ->
+          query
+
         {:connection, _value} ->
           where(query, [d], d.connection_status == ^String.to_atom(value))
 

lib/nerves_hub/devices/device.ex

@@ -5,6 +5,7 @@ defmodule NervesHub.Devices.Device do
 
   alias NervesHub.Accounts.Org
   alias NervesHub.Devices.DeviceCertificate
+  alias NervesHub.Devices.DeviceMetric
   alias NervesHub.Deployments.Deployment
   alias NervesHub.Firmwares.FirmwareMetadata
   alias NervesHub.Products.Product
@@ -36,6 +37,7 @@ defmodule NervesHub.Devices.Device do
     belongs_to(:deployment, Deployment)
     embeds_one(:firmware_metadata, FirmwareMetadata, on_replace: :update)
     has_many(:device_certificates, DeviceCertificate, on_delete: :delete_all)
+    has_many(:device_metrics, DeviceMetric, on_delete: :delete_all)
 
     field(:identifier, :string)
     field(:description, :string)

lib/nerves_hub_web/live/devices/index.ex

@@ -30,7 +30,8 @@ defmodule NervesHubWeb.Live.Devices.Index do
     updates: "",
     has_no_tags: false,
     alarm_status: "",
-    alarm: ""
+    alarm: "",
+    metrics: %{}
   }
 
   @filter_types %{
@@ -44,7 +45,8 @@ defmodule NervesHubWeb.Live.Devices.Index do
     updates: :string,
     has_no_tags: :boolean,
     alarm_status: :string,
-    alarm: :string
+    alarm: :string,
+    metrics: :map
   }
 
   @default_page 1
@@ -88,7 +90,11 @@ defmodule NervesHubWeb.Live.Devices.Index do
   end
 
   def handle_params(unsigned_params, _uri, socket) do
-    filters = Map.merge(@default_filters, filter_changes(unsigned_params))
+    filters =
+      Map.merge(@default_filters, filter_changes(unsigned_params))
+      |> filter_on_metrics_if_provided(unsigned_params)
+      |> dbg()
+
     pagination_opts = Map.merge(socket.assigns.paginate_opts, pagination_changes(unsigned_params))
 
     socket
@@ -102,6 +108,22 @@ defmodule NervesHubWeb.Live.Devices.Index do
     |> noreply()
   end
 
+  def filter_on_metrics_if_provided(filters, %{
+        "metric" => metric_type,
+        "metric_value" => metric_value,
+        "metric_operator" => operator
+      }) do
+    Map.put(filters, :metrics, %{
+      key: metric_type,
+      value: String.to_float(metric_value),
+      operator: String.to_existing_atom(operator)
+    })
+  end
+
+  def filter_on_metrics_if_provided(filters, _) do
+    filters
+  end
+
   defp self_path(socket, extra) do
     params = Enum.into(stringify_keys(extra), socket.assigns.params)
     pagination = pagination_changes(params)