wip

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

@@ -32,23 +32,28 @@ 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)}")
 
+    # TODO: this needs to broadcast either a cluster execution started or a host execution started
     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 <-
+    with "cluster" <- target_type,
+         :ok <-
            commanded().dispatch(
              CompleteChecksExecution.new!(%{
                cluster_id: group_id,
@@ -57,8 +62,14 @@ defmodule Trento.Integration.Checks.AMQP.Processor do
              correlation_id: execution_id
            ) do
       TrentoWeb.Endpoint.broadcast("monitoring:executions", "execution_completed", %{
-        group_id: group_id
+        group_id: group_id,
+        target_type: target_type
       })
+    else
+      _zs ->
+        IO.inspect(_zs)
+        # host execution completion to be implemented
+        :ok
     end
   end
 

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

@@ -3,6 +3,7 @@ defmodule Trento.Integration.Checks do
   Checks runner service integration
   """
 
+  alias Trento.HostReadModel
   alias Trento.Infrastructure.Messaging
 
   alias Trento.Checks.V1.{
@@ -11,20 +12,53 @@ defmodule Trento.Integration.Checks do
   }
 
   alias Trento.Integration.Checks.ClusterExecutionEnv
+  alias Trento.Integration.Checks.HostExecutionEnv
 
   require Logger
 
-  @spec request_execution(String.t(), String.t(), ClusterExecutionEnv.t(), [map], [String.t()]) ::
+  @type target_type :: :host | :cluster
+
+  @supported_targets [
+    :host,
+    :cluster
+  ]
+
+  @spec request_host_execution(
+          String.t(),
+          String.t(),
+          HostExecutionEnv.t(),
+          [String.t()]
+        ) ::
+          :ok | {:error, :any}
+  def request_host_execution(execution_id, host_id, env, selected_checks) do
+    request_execution(execution_id, host_id, env, [%{host_id: host_id}], selected_checks, :host)
+  end
+
+  @spec request_cluster_execution(
+          String.t(),
+          String.t(),
+          ClusterExecutionEnv.t(),
+          [HostReadModel.t()],
+          [String.t()]
+        ) ::
           :ok | {:error, :any}
-  def request_execution(execution_id, cluster_id, env, hosts, selected_checks) do
+  def request_cluster_execution(execution_id, cluster_id, env, hosts, selected_checks) do
+    request_execution(execution_id, cluster_id, env, hosts, selected_checks, :cluster)
+  end
+
+  defp request_execution(execution_id, target_id, env, targets, selected_checks, target_type) 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: Atom.to_string(target_type)
     }
 
     case Messaging.publish("executions", execution_requested) do
@@ -44,4 +78,10 @@ defmodule Trento.Integration.Checks do
       "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
 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/usecases/clusters/clusters.ex

@@ -144,7 +144,7 @@ defmodule Trento.Clusters do
           where: h.cluster_id == ^cluster_id and is_nil(h.deregistered_at)
       )
 
-    Checks.request_execution(
+    Checks.request_cluster_execution(
       UUID.uuid4(),
       cluster_id,
       %Checks.ClusterExecutionEnv{

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,43 @@ 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
+
+  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_host_execution(
+      UUID.uuid4(),
+      host_id,
+      %Checks.HostExecutionEnv{provider: provider},
+      selected_checks
+    )
+  end
+
+  # defp maybe_request_checks_execution(error), do: error
+
   @spec deregister_host(Ecto.UUID.t(), DateService) ::
           :ok | {:error, :host_alive} | {:error, :host_not_registered}
   def deregister_host(host_id, date_service \\ DateService) 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,30 @@ 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(),
+      bad_request: Schema.BadRequest.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/trento/application/integration/checks/checks_test.exs

@@ -46,7 +46,7 @@ defmodule Trento.Integration.ChecksTest do
     end)
 
     assert :ok =
-             Checks.request_execution(
+             Checks.request_cluster_execution(
                execution_id,
                group_id,
                env,

test/trento/application/integration/checks/processor_test.exs

@@ -68,7 +68,8 @@ defmodule Trento.Integration.Checks.AMQP.ProcessorTest do
         Contracts.to_event(%ExecutionCompleted{
           execution_id: execution_id,
           group_id: group_id,
-          result: :PASSING
+          result: :PASSING,
+          target_type: "cluster"
         })
 
       message = %GenRMQ.Message{payload: execution_completed, attributes: %{}, channel: nil}
@@ -94,7 +95,8 @@ defmodule Trento.Integration.Checks.AMQP.ProcessorTest do
         Contracts.to_event(%ExecutionCompleted{
           execution_id: UUID.uuid4(),
           group_id: group_id,
-          result: :PASSING
+          result: :PASSING,
+          target_type: "cluster"
         })
 
       message = %GenRMQ.Message{payload: execution_completed, attributes: %{}, channel: nil}