Host checks execution api

lib/trento/application/event_handlers/stream_roll_up_event_handler.ex

@@ -45,7 +45,8 @@ defmodule Trento.StreamRollUpEventHandler do
     Trento.Domain.Events.HostDetailsUpdated,
     Trento.Domain.Events.HostRegistered,
     Trento.Domain.Events.ProviderUpdated,
-    Trento.Domain.Events.SlesSubscriptionsUpdated
+    Trento.Domain.Events.SlesSubscriptionsUpdated,
+    Trento.Domain.Events.HostChecksSelected
   ]
 
   @sap_system_events [

lib/trento/application/integration/checks/amqp/processor.ex

@@ -12,7 +12,8 @@ defmodule Trento.Integration.Checks.AMQP.Processor do
     ExecutionStarted
   }
 
-  alias Trento.Domain.Commands.CompleteChecksExecution
+  alias Trento.Integration.Checks
+  alias Trento.Integration.Checks.TargetType
 
   require Logger
   require Trento.Domain.Enums.Health, as: Health
@@ -32,32 +33,35 @@ defmodule Trento.Integration.Checks.AMQP.Processor do
   defp handle(%ExecutionStarted{
          execution_id: execution_id,
          group_id: group_id,
-         targets: targets
+         targets: targets,
+         target_type: target_type
        }) do
     Logger.debug("Targets for execution #{inspect(targets)}")
 
     TrentoWeb.Endpoint.broadcast("monitoring:executions", "execution_started", %{
       group_id: group_id,
       execution_id: execution_id,
-      targets: map_targets(targets)
+      targets: map_targets(targets),
+      target_type: target_type
     })
   end
 
   defp handle(%ExecutionCompleted{
          execution_id: execution_id,
          group_id: group_id,
-         result: result
+         result: result,
+         target_type: target_type
        }) do
     with :ok <-
-           commanded().dispatch(
-             CompleteChecksExecution.new!(%{
-               cluster_id: group_id,
-               health: map_health(result)
-             }),
-             correlation_id: execution_id
+           Checks.complete_execution(
+             execution_id,
+             group_id,
+             map_health(result),
+             TargetType.from_string(target_type)
            ) do
       TrentoWeb.Endpoint.broadcast("monitoring:executions", "execution_completed", %{
-        group_id: group_id
+        group_id: group_id,
+        target_type: target_type
       })
     end
   end
@@ -67,7 +71,4 @@ defmodule Trento.Integration.Checks.AMQP.Processor do
   defp map_health(:CRITICAL), do: Health.critical()
   defp map_health(:WARNING), do: Health.warning()
   defp map_health(:PASSING), do: Health.passing()
-
-  defp commanded,
-    do: Application.fetch_env!(:trento, Trento.Commanded)[:adapter]
 end

lib/trento/application/integration/checks/checks.ex

@@ -1,30 +1,54 @@
 defmodule Trento.Integration.Checks do
   @moduledoc """
-  Checks runner service integration
+  Checks Engine service integration
   """
 
+  alias Trento.Domain.Commands.CompleteChecksExecution
+  alias Trento.Domain.Enums.Health
   alias Trento.Infrastructure.Messaging
 
   alias Trento.Checks.V1.{
     ExecutionRequested,
     Target
   }
 
-  alias Trento.Integration.Checks.ClusterExecutionEnv
+  alias Trento.Integration.Checks.{
+    ClusterExecutionEnv,
+    HostExecutionEnv
+  }
 
   require Logger
+  require Trento.Integration.Checks.TargetType, as: TargetType
+
+  @type target_type :: TargetType.t()
+  @type target_env :: HostExecutionEnv.t() | ClusterExecutionEnv.t()
+  @type targets :: [%{host_id: String.t()}]
 
-  @spec request_execution(String.t(), String.t(), ClusterExecutionEnv.t(), [map], [String.t()]) ::
-          :ok | {:error, :any}
-  def request_execution(execution_id, cluster_id, env, hosts, selected_checks) do
+  @supported_targets TargetType.values()
+
+  @spec request_execution(
+          String.t(),
+          String.t(),
+          target_env,
+          targets,
+          [String.t()],
+          target_type
+        ) ::
+          :ok | {:error, any}
+  def request_execution(execution_id, target_id, env, targets, selected_checks, target_type)
+      when target_type in @supported_targets do
     execution_requested = %ExecutionRequested{
       execution_id: execution_id,
-      group_id: cluster_id,
+      group_id: target_id,
       targets:
-        Enum.map(hosts, fn %{host_id: host_id} ->
-          %Target{agent_id: host_id, checks: selected_checks}
-        end),
-      env: build_env(env)
+        Enum.map(
+          targets,
+          fn %{host_id: host_id} ->
+            %Target{agent_id: host_id, checks: selected_checks}
+          end
+        ),
+      env: build_env(env),
+      target_type: TargetType.to_string(target_type)
     }
 
     case Messaging.publish("executions", execution_requested) do
@@ -38,10 +62,39 @@ defmodule Trento.Integration.Checks do
     end
   end
 
+  def request_execution(_, _, _, _, _, _), do: {:error, :target_not_supported}
+
+  @spec complete_execution(String.t(), String.t(), Health.t(), target_type) :: :ok | {:error, any}
+  def complete_execution(execution_id, target_id, health, :cluster) do
+    commanded().dispatch(
+      CompleteChecksExecution.new!(%{
+        cluster_id: target_id,
+        health: health
+      }),
+      correlation_id: execution_id
+    )
+  end
+
+  def complete_execution(_execution_id, _target_id, _health, :host) do
+    # TODO dispatch host execution completed command
+    :ok
+  end
+
+  def complete_execution(_, _, _, _), do: {:error, :target_not_supported}
+
   defp build_env(%ClusterExecutionEnv{cluster_type: cluster_type, provider: provider}) do
     %{
       "cluster_type" => %{kind: {:string_value, Atom.to_string(cluster_type)}},
       "provider" => %{kind: {:string_value, Atom.to_string(provider)}}
     }
   end
+
+  defp build_env(%HostExecutionEnv{provider: provider}) do
+    %{
+      "provider" => %{kind: {:string_value, Atom.to_string(provider)}}
+    }
+  end
+
+  defp commanded,
+    do: Application.fetch_env!(:trento, Trento.Commanded)[:adapter]
 end

lib/trento/application/integration/checks/host_executions_env.ex

@@ -0,0 +1,14 @@
+defmodule Trento.Integration.Checks.HostExecutionEnv do
+  @moduledoc """
+  Host checks execution env map
+  """
+
+  @required_fields :all
+  use Trento.Type
+
+  require Trento.Domain.Enums.Provider, as: Provider
+
+  deftype do
+    field :provider, Ecto.Enum, values: Provider.values()
+  end
+end

lib/trento/application/integration/checks/target_type.ex

@@ -0,0 +1,15 @@
+defmodule Trento.Integration.Checks.TargetType do
+  @moduledoc """
+  Type that represents the possible target types for a check execution.
+  """
+
+  use Trento.Support.Enum, values: [:cluster, :host]
+
+  def from_string("cluster"), do: cluster()
+  def from_string("host"), do: host()
+  def from_string(_), do: nil
+
+  def to_string(cluster()), do: "cluster"
+  def to_string(host()), do: "host"
+  def to_string(_), do: nil
+end

lib/trento/application/usecases/clusters/clusters.ex

@@ -152,7 +152,8 @@ defmodule Trento.Clusters do
         cluster_type: cluster_type
       },
       hosts_data,
-      selected_checks
+      selected_checks,
+      :cluster
     )
   end
 

lib/trento/application/usecases/hosts/hosts.ex

@@ -21,6 +21,8 @@ defmodule Trento.Hosts do
     SelectHostChecks
   }
 
+  alias Trento.Integration.Checks
+
   alias Trento.Repo
 
   @spec get_all_hosts :: [HostReadModel.t()]
@@ -67,6 +69,26 @@ defmodule Trento.Hosts do
     end
   end
 
+  @spec request_checks_execution(String.t()) :: :ok | {:error, any}
+  def request_checks_execution(host_id) do
+    query =
+      from(h in HostReadModel,
+        where: is_nil(h.deregistered_at) and h.id == ^host_id
+      )
+
+    case Repo.one(query) do
+      %HostReadModel{} = host ->
+        Logger.debug("Requesting checks execution, host: #{host_id}")
+
+        maybe_request_checks_execution(host)
+
+      nil ->
+        Logger.error("Requested checks execution for a non-existing host: #{host_id}")
+
+        {:error, :not_found}
+    end
+  end
+
   @spec deregister_host(Ecto.UUID.t(), DateService) ::
           :ok | {:error, :host_alive} | {:error, :host_not_registered}
   def deregister_host(host_id, date_service \\ DateService) do
@@ -82,6 +104,23 @@ defmodule Trento.Hosts do
     |> select_merge([h, hb], %{last_heartbeat_timestamp: hb.timestamp})
   end
 
+  defp maybe_request_checks_execution(%{selected_checks: []}), do: {:error, :no_checks_selected}
+
+  defp maybe_request_checks_execution(%{
+         id: host_id,
+         selected_checks: selected_checks,
+         provider: provider
+       }) do
+    Checks.request_execution(
+      UUID.uuid4(),
+      host_id,
+      %Checks.HostExecutionEnv{provider: provider},
+      [%{host_id: host_id}],
+      selected_checks,
+      :host
+    )
+  end
+
   defp commanded,
     do: Application.fetch_env!(:trento, Trento.Commanded)[:adapter]
 end

lib/trento_web/controllers/fallback_controller.ex

@@ -74,6 +74,13 @@ defmodule TrentoWeb.FallbackController do
     |> render(:"422", reason: "Requested operation not allowed for present SAP instances.")
   end
 
+  def call(conn, {:error, :no_checks_selected}) do
+    conn
+    |> put_status(:unprocessable_entity)
+    |> put_view(ErrorView)
+    |> render(:"422", reason: "No checks were selected for the target.")
+  end
+
   def call(conn, {:error, [error | _]}), do: call(conn, {:error, error})
 
   def call(conn, {:error, _}) do

lib/trento_web/controllers/v1/cluster_controller.ex

@@ -37,7 +37,7 @@ defmodule TrentoWeb.V1.ClusterController do
       ]
     ],
     responses: [
-      accepted: "The Command has been accepted and the Requested execution is scheduled",
+      accepted: "The Command has been accepted and the Requested Cluster execution is scheduled",
       not_found: Schema.NotFound.response(),
       bad_request: Schema.BadRequest.response(),
       unprocessable_entity: OpenApiSpex.JsonErrorResponse.response()

lib/trento_web/controllers/v1/host_controller.ex

@@ -109,4 +109,29 @@ defmodule TrentoWeb.V1.HostController do
       |> json(%{})
     end
   end
+
+  operation :request_checks_execution,
+    summary: "Request Checks Execution for a Host",
+    tags: ["Checks"],
+    description: "Trigger execution of the latest Checks Selection on the target infrastructure",
+    parameters: [
+      id: [
+        in: :path,
+        required: true,
+        type: %OpenApiSpex.Schema{type: :string, format: :uuid}
+      ]
+    ],
+    responses: [
+      accepted: "The Command has been accepted and the Requested Host execution is scheduled",
+      not_found: Schema.NotFound.response(),
+      unprocessable_entity: OpenApiSpex.JsonErrorResponse.response()
+    ]
+
+  def request_checks_execution(conn, %{id: host_id}) do
+    with :ok <- Hosts.request_checks_execution(host_id) do
+      conn
+      |> put_status(:accepted)
+      |> json(%{})
+    end
+  end
 end

lib/trento_web/router.ex

@@ -97,6 +97,10 @@ defmodule TrentoWeb.Router do
            ClusterController,
            :request_checks_execution
 
+      post "/hosts/:id/checks/request_execution",
+           HostController,
+           :request_checks_execution
+
       post "/hosts/:id/tags", TagsController, :add_tag,
         assigns: %{resource_type: :host},
         as: :hosts_tagging

test/support/factory.ex

@@ -112,7 +112,8 @@ defmodule Trento.Factory do
       heartbeat: :unknown,
       provider: Enum.random(Provider.values()),
       provider_data: nil,
-      deregistered_at: nil
+      deregistered_at: nil,
+      selected_checks: Enum.map(0..4, fn _ -> Faker.StarWars.planet() end)
     }
   end