feat: add runtime liveness reporting

Author: Bolyki György

lib/tesla_api/stream.ex

@@ -11,6 +11,7 @@ defmodule TeslaApi.Stream do
               vehicle_id: nil,
               timer: nil,
               receiver: &IO.inspect/1,
+              liveness_receiver: nil,
               last_data: nil,
               timeouts: 0,
               disconnects: 0
@@ -27,6 +28,7 @@ defmodule TeslaApi.Stream do
   def start_link(args) do
     state = %State{
       receiver: Keyword.get(args, :receiver, &Logger.debug(inspect(&1))),
+      liveness_receiver: Keyword.get(args, :liveness_receiver, fn _event -> :ok end),
       vehicle_id: Keyword.fetch!(args, :vehicle_id),
       auth: Keyword.fetch!(args, :auth)
     }
@@ -56,13 +58,15 @@ defmodule TeslaApi.Stream do
 
   @impl true
   def handle_cast(:disconnect, %State{vehicle_id: vid} = state) do
+    notify_liveness(state, :disconnected)
     send(self(), :exit)
     {:reply, frame!(%{msg_type: "data:unsubscribe", tag: "#{vid}"}), state}
   end
 
   @impl true
-  def handle_connect(_conn, state) do
+  def handle_connect(_conn, %State{} = state) do
     Logger.debug("Connection established")
+    notify_liveness(state, :connected)
     send(self(), :subscribe)
     {:ok, state}
   end
@@ -203,13 +207,16 @@ defmodule TeslaApi.Stream do
 
     case reason do
       {:local, :normal} ->
+        notify_liveness(state, :reconnecting)
+
         Logger.debug(
           "Connection was closed (a:#{n}|t:#{state.timeouts}|d:#{state.disconnects}). Reconnecting …"
         )
 
         {:reconnect, state}
 
       {:remote, :closed} ->
+        notify_liveness(state, :reconnecting)
         Logger.warning("WebSocket disconnected. Reconnecting …")
 
         n
@@ -219,6 +226,7 @@ defmodule TeslaApi.Stream do
         {:reconnect, %{state | last_data: nil}}
 
       %WebSockex.ConnError{} = e ->
+        notify_liveness(state, :reconnecting)
         Logger.warning("Disconnected! #{Exception.message(e)} | #{n}")
 
         n
@@ -228,6 +236,7 @@ defmodule TeslaApi.Stream do
         {:reconnect, state}
 
       %WebSockex.RequestError{} = e ->
+        notify_liveness(state, :reconnecting)
         Logger.warning("Disconnected! #{Exception.message(e)} | #{n}")
 
         n
@@ -241,7 +250,9 @@ defmodule TeslaApi.Stream do
   @impl true
   def terminate(:normal, _state), do: :ok
 
-  def terminate(reason, _state) do
+  def terminate(reason, %State{} = state) do
+    notify_liveness(state, :terminated)
+
     # https://github.com/Azolo/websockex/issues/51
     with {exception, stacktrace} <- reason, true <- is_exception(exception) do
       Logger.error(fn -> Exception.format(:error, exception, stacktrace) end)
@@ -264,6 +275,9 @@ defmodule TeslaApi.Stream do
     :math.pow(base, n) |> min(max) |> max(min) |> round() |> :timer.seconds()
   end
 
+  defp notify_liveness(%State{liveness_receiver: nil}, _event), do: :ok
+  defp notify_liveness(%State{liveness_receiver: receiver}, event), do: receiver.(event)
+
   defp cancel_timer(nil), do: :ok
   defp cancel_timer(ref) when is_reference(ref), do: Process.cancel_timer(ref)
 end

lib/teslamate/api.ex

@@ -49,9 +49,26 @@ defmodule TeslaMate.Api do
     end
   end
 
-  def stream(name \\ @name, vid, receiver) do
+  def stream(vid, receiver) when is_function(receiver, 1) do
+    stream(@name, vid, receiver, [])
+  end
+
+  def stream(vid, receiver, opts) when is_function(receiver, 1) and is_list(opts) do
+    stream(@name, vid, receiver, opts)
+  end
+
+  def stream(name, vid, receiver) when is_function(receiver, 1) do
+    stream(name, vid, receiver, [])
+  end
+
+  def stream(name, vid, receiver, opts) when is_function(receiver, 1) and is_list(opts) do
     with {:ok, %Auth{} = auth} <- fetch_auth(name) do
-      TeslaApi.Stream.start_link(auth: auth, vehicle_id: vid, receiver: receiver)
+      TeslaApi.Stream.start_link(
+        auth: auth,
+        vehicle_id: vid,
+        receiver: receiver,
+        liveness_receiver: Keyword.get(opts, :liveness_receiver, fn _event -> :ok end)
+      )
     end
   end
 

lib/teslamate/application.ex

@@ -28,6 +28,7 @@ defmodule TeslaMate.Application do
           TeslaMate.Api,
           TeslaMate.Updater,
           {Phoenix.PubSub, name: TeslaMate.PubSub},
+          {TeslaMate.RuntimeHealth, mqtt_enabled: mqtt_config != nil},
           TeslaMateWeb.Endpoint,
           TeslaMate.Terrain,
           TeslaMate.Vehicles,
@@ -44,6 +45,7 @@ defmodule TeslaMate.Application do
           TeslaMate.Api,
           TeslaMate.Updater,
           {Phoenix.PubSub, name: TeslaMate.PubSub},
+          {TeslaMate.RuntimeHealth, mqtt_enabled: false},
           TeslaMateWeb.Endpoint,
           {TeslaMate.Terrain, disabled: true},
           {TeslaMate.Repair, limit: 250},

lib/teslamate/import/fake_api.ex

@@ -33,6 +33,10 @@ defmodule TeslaMate.Import.FakeApi do
     {:ok, nil}
   end
 
+  def stream(_name, _vid, _receiver, _opts) do
+    {:ok, nil}
+  end
+
   # Callbacks
 
   @impl true

lib/teslamate/mqtt.ex

@@ -53,7 +53,7 @@ defmodule TeslaMate.Mqtt do
       user_name: opts[:username],
       password: opts[:password],
       server: server,
-      handler: {Handler, []},
+      handler: {Handler, [runtime_health: TeslaMate.RuntimeHealth]},
       subscriptions: []
     ]
   end

lib/teslamate/mqtt/handler.ex

@@ -2,26 +2,39 @@ defmodule TeslaMate.Mqtt.Handler do
   use Tortoise311.Handler
 
   require Logger
+  import Core.Dependency, only: [call: 3]
+
+  alias TeslaMate.RuntimeHealth
 
   @impl true
   def connection(:up, state) do
     Logger.info("MQTT connection has been established")
+    :ok = call(runtime_health(state), :record_mqtt_connection, [:up])
     {:ok, state}
   end
 
   def connection(:down, state) do
     Logger.warning("MQTT connection has been dropped")
+    :ok = call(runtime_health(state), :record_mqtt_connection, [:down])
     {:ok, state}
   end
 
   def connection(:terminating, state) do
     Logger.warning("MQTT connection is terminating")
+    :ok = call(runtime_health(state), :record_mqtt_connection, [:terminating])
     {:ok, state}
   end
 
   @impl true
-  def terminate(reason, _state) do
+  def terminate(reason, state) do
     Logger.warning("MQTT Client has been terminated with reason: #{inspect(reason)}")
+    :ok = call(runtime_health(state), :record_mqtt_connection, [:terminated])
     :ok
   end
+
+  defp runtime_health(state) when is_list(state) do
+    Keyword.get(state, :runtime_health, RuntimeHealth)
+  end
+
+  defp runtime_health(_state), do: RuntimeHealth
 end

lib/teslamate/mqtt/pubsub/vehicle_subscriber.ex

@@ -5,10 +5,11 @@ defmodule TeslaMate.Mqtt.PubSub.VehicleSubscriber do
   import Core.Dependency, only: [call: 3]
 
   alias TeslaMate.Mqtt.Publisher
+  alias TeslaMate.RuntimeHealth
   alias TeslaMate.Vehicles.Vehicle.Summary
   alias TeslaMate.Vehicles
 
-  defstruct [:car_id, :last_values, :deps, :namespace]
+  defstruct [:car_id, :last_values, :deps, :namespace, mqtt_generation: 0]
   alias __MODULE__, as: State
 
   def child_spec(arg) do
@@ -45,13 +46,31 @@ defmodule TeslaMate.Mqtt.PubSub.VehicleSubscriber do
 
     deps = %{
       vehicles: Keyword.get(opts, :deps_vehicles, Vehicles),
-      publisher: Keyword.get(opts, :deps_publisher, Publisher)
+      publisher: Keyword.get(opts, :deps_publisher, Publisher),
+      runtime_health: Keyword.get(opts, :deps_runtime_health, RuntimeHealth)
     }
 
     :ok = call(deps.vehicles, :subscribe_to_summary, [car_id])
+    :ok = call(deps.runtime_health, :subscribe_mqtt, [])
     :ok = clear_retained(car_id, namespace, deps.publisher)
 
-    {:ok, %State{car_id: car_id, namespace: namespace, deps: deps}}
+    mqtt_snapshot = call(deps.runtime_health, :mqtt_snapshot, [])
+
+    mqtt_generation =
+      if mqtt_snapshot.status == :ok do
+        send(self(), {:mqtt_reconnected, mqtt_snapshot.generation})
+        nil
+      else
+        mqtt_snapshot.generation
+      end
+
+    {:ok,
+     %State{
+       car_id: car_id,
+       namespace: namespace,
+       deps: deps,
+       mqtt_generation: mqtt_generation
+     }}
   end
 
   @publish_if_nil ~w(charge_energy_added charger_actual_current charger_phases
@@ -60,14 +79,36 @@ defmodule TeslaMate.Mqtt.PubSub.VehicleSubscriber do
 
   @impl true
   def handle_info(%Summary{} = summary, %State{} = state) do
+    state = reconcile_mqtt_generation(state)
+    publish_summary(summary, state)
+  end
+
+  def handle_info({:mqtt_reconnected, generation}, %State{} = state) do
+    mqtt_snapshot = current_mqtt_snapshot(state)
+
+    if mqtt_snapshot.status == :ok and mqtt_snapshot.generation == generation and
+         generation != state.mqtt_generation do
+      state = %{state | last_values: nil, mqtt_generation: generation}
+
+      case current_summary(state) do
+        %Summary{} = summary -> publish_summary(summary, state)
+        _other -> {:noreply, state}
+      end
+    else
+      {:noreply, state}
+    end
+  end
+
+  defp publish_summary(%Summary{} = summary, %State{} = state) do
     values =
       %{}
       |> add_simple_values(summary)
       |> add_car_latitude_longitude(summary)
       |> add_geofence(summary)
       |> add_active_route(summary)
 
-    last_values = publish_values(values, state)
+    {last_values, result} = publish_values(values, state)
+    :ok = call(state.deps.runtime_health, :record_mqtt_publish, [state.car_id, result])
     {:noreply, %{state | last_values: last_values}}
   end
 
@@ -86,17 +127,21 @@ defmodule TeslaMate.Mqtt.PubSub.VehicleSubscriber do
       on_timeout: :kill_task,
       ordered: false
     )
-    |> Enum.reduce(last_values, fn
-      {:ok, {{key, value}, :ok}}, acc ->
-        Map.put(acc, key, value)
+    |> Enum.reduce({last_values, 0}, fn
+      {:ok, {{key, value}, :ok}}, {acc, failures} ->
+        {Map.put(acc, key, value), failures}
 
-      {:ok, {_entry, reason}}, acc ->
+      {:ok, {_entry, reason}}, {acc, failures} ->
         Logger.warning("MQTT publishing failed: #{inspect(reason)}")
-        acc
+        {acc, failures + 1}
 
-      {:exit, reason}, acc ->
+      {:exit, reason}, {acc, failures} ->
         Logger.warning("MQTT publishing failed: #{inspect(reason)}")
-        acc
+        {acc, failures + 1}
+    end)
+    |> then(fn
+      {values, 0} -> {values, :ok}
+      {values, _failures} -> {values, {:error, :publish_failed}}
     end)
   end
 
@@ -214,4 +259,29 @@ defmodule TeslaMate.Mqtt.PubSub.VehicleSubscriber do
 
   defp to_str(%DateTime{} = datetime), do: DateTime.to_iso8601(datetime)
   defp to_str(value), do: to_string(value)
+
+  defp current_summary(%State{car_id: car_id, deps: deps}) do
+    call(deps.vehicles, :summary, [car_id])
+  catch
+    :exit, reason ->
+      Logger.warning("Could not republish MQTT snapshot: #{inspect(reason)}")
+      nil
+  end
+
+  defp reconcile_mqtt_generation(%State{} = state) do
+    case current_mqtt_snapshot(state) do
+      %{status: :ok, generation: generation}
+      when not is_nil(generation) and generation != state.mqtt_generation ->
+        %{state | last_values: nil, mqtt_generation: generation}
+
+      _snapshot ->
+        state
+    end
+  end
+
+  defp current_mqtt_snapshot(%State{deps: deps}) do
+    call(deps.runtime_health, :mqtt_snapshot, [])
+  catch
+    :exit, _reason -> %{status: :unavailable, generation: nil}
+  end
 end