Add alert sideloading

Needed to help with ai insight sideload needs

assets/src/generated/graphql.ts

@@ -4849,6 +4849,8 @@ export type PolicyEngine = {
 };
 
 export type PolicyEngineAttributes = {
+  /** the maximum allowed severity without failing the stack run */
+  maxSeverity?: InputMaybe<VulnSeverity>;
   /** the policy engine to use with this stack */
   type: PolicyEngineType;
 };
@@ -7111,6 +7113,7 @@ export type RootQueryTypeClustersArgs = {
   q?: InputMaybe<Scalars['String']['input']>;
   tag?: InputMaybe<TagInput>;
   tagQuery?: InputMaybe<TagQuery>;
+  upgradeable?: InputMaybe<Scalars['Boolean']['input']>;
 };
 
 
@@ -8964,13 +8967,15 @@ export enum StackType {
 export type StackViolationCause = {
   __typename?: 'StackViolationCause';
   end: Scalars['Int']['output'];
+  filename?: Maybe<Scalars['String']['output']>;
   lines?: Maybe<Array<Maybe<StackViolationCauseLine>>>;
   resource: Scalars['String']['output'];
   start: Scalars['Int']['output'];
 };
 
 export type StackViolationCauseAttributes = {
   end: Scalars['Int']['input'];
+  filename?: InputMaybe<Scalars['String']['input']>;
   lines?: InputMaybe<Array<InputMaybe<StackViolationCauseLineAttributes>>>;
   resource: Scalars['String']['input'];
   start: Scalars['Int']['input'];

config/config.exs

@@ -112,7 +112,7 @@ config :console, Console.PartitionedCache,
     allocated_memory: 1000 * 1000 * 500
   ]
 
-config :my_app, Console.MultilevelCache,
+config :console, Console.MultilevelCache,
   model: :inclusive,
   levels: [
     {
@@ -161,4 +161,8 @@ config :console, Console.PromEx,
   metrics_server: :disabled,
   version: "0.11.0"
 
+config :console, :ttls,
+  helm: :timer.minutes(30),
+  cluster_metrics: :timer.hours(6)
+
 import_config "#{Mix.env()}.exs"

config/prod.exs

@@ -41,6 +41,7 @@ config :console, Console.Cron.Scheduler,
     {"*/4 * * * *",    {Console.Deployments.Cron, :scan_pipeline_stages, []}},
     {"*/4 * * * *",    {Console.Deployments.Cron, :scan_pending_promotions, []}},
     {"*/4 * * * *",    {Console.Deployments.Cron, :scan_pending_contexts, []}},
+    {"*/5 * * * *",    {Console.Deployments.Cron, :cache_warm, []}},
     {"*/10 * * * *",   {Console.Deployments.Init, :ensure_secret, []}},
     {"*/5 * * * *",    {Console.AI.Cron, :services, []}},
     {"*/5 * * * *",    {Console.AI.Cron, :stacks, []}},

lib/console/application.ex

@@ -14,13 +14,15 @@ defmodule Console.Application do
       {Phoenix.PubSub, [name: Console.PubSub, adapter: Phoenix.PubSub.PG2]},
       Console.Cache,
       Console.ReplicatedCache,
+      Console.MultilevelCache,
       Console.TestCache,
       Console.LocalCache,
       ConsoleWeb.Endpoint,
       Console.Plural.Config,
       Console.Features,
       Console.Cron.Scheduler,
       Console.Buffers.ClusterAudit,
+      Console.Deployments.Local.Server,
       {Registry, [keys: :unique, name: Console.Buffer.Base.registry()]},
       {Registry, [keys: :unique, name: Console.Deployments.Git.Agent.registry()]},
       {Registry, [keys: :unique, name: Console.Deployments.Pipelines.Supervisor.registry()]},

lib/console/cached/cluster_nodes.ex

@@ -2,33 +2,38 @@ defmodule Console.Cached.ClusterNodes do
   @moduledoc "this will perpertually warm the nebulex cache for cluster nodes locally"
   use GenServer
   require Logger
-  alias Console.Deployments.{Cron}
 
-  @warm :timer.minutes(5)
+  @metrics_ttl Console.conf(:ttls)[:cluster_metrics]
+  @helm_ttl Console.conf(:ttls)[:helm]
+  @local_adapter Console.conf(:local_cache)
 
   def start_link(opt \\ :ok) do
     GenServer.start_link(__MODULE__, opt, name: __MODULE__)
   end
 
+  def broadcast(msg) do
+    :pg.get_members(__MODULE__)
+    |> Enum.each(&GenServer.cast(&1, {:broadcast, msg}))
+  end
+
+  def cluster_metrics(id, metrics), do: broadcast({:cluster_metrics, id, metrics})
+
+  def helm_repos(repos), do: broadcast({:helm_repos, repos})
+
   def init(_) do
-    if Console.conf(:initialize) do
-      :timer.send_interval(@warm, :warm)
-      send self(), :warm
-    end
+    :pg.join(__MODULE__, self())
     {:ok, %{}}
   end
 
-  def handle_info(:warm, s) do
-    Logger.info "warming cluster info caches"
-    try do
-      Cron.cache_warm()
-    rescue
-      e ->
-        Logger.info "hit error trying to warm node caches"
-        Logger.error(Exception.format(:error, e, __STACKTRACE__))
-    end
+  def handle_cast({:broadcast, {:cluster_metrics, id, metrics}}, s) do
+    @local_adapter.put({:cluster_metrics, id}, {:ok, metrics}, ttl: @metrics_ttl)
+    {:noreply, s}
+  end
+
+  def handle_cast({:broadcast, {:helm_repos, repos}}, s) do
+    @local_adapter.put(:helm_repositories, repos, ttl: @helm_ttl)
     {:noreply, s}
   end
 
-  def handle_info(_, s), do: {:noreply, s}
+  def handle_cast(_, s), do: {:noreply, s}
 end

lib/console/deployments/clusters.ex

@@ -9,6 +9,7 @@ defmodule Console.Deployments.Clusters do
   alias Console.Deployments.Compatibilities.{Table, AddOn, Version, CloudAddOns}
   alias Console.Deployments.Ecto.Validations
   alias Console.Services.Users
+  alias Console.Cached.ClusterNodes
   alias Kazan.Apis.Core.V1, as: Core
   alias Console.Schema.{
     Cluster,
@@ -169,7 +170,7 @@ defmodule Console.Deployments.Clusters do
 
   def warm(:cluster_metrics, %Cluster{id: id} = cluster) do
     with {:ok, metrics} <- fetch_cluster_metrics(cluster),
-      do: @local_adapter.put({:cluster_metrics, id}, {:ok, metrics}, ttl: @node_ttl)
+      do: ClusterNodes.cluster_metrics(id, metrics)
   end
 
   def warm(:api_discovery, %Cluster{} = cluster), do: api_discovery(cluster)

lib/console/deployments/cron.ex

@@ -1,6 +1,6 @@
 defmodule Console.Deployments.Cron do
   use Console.Services.Base
-  import Console, only: [clamp: 1]
+  import Console, only: [clamp: 1, clamp: 3]
   alias Console.Deployments.{Services, Clusters, Global, Stacks, Git}
   alias Console.Services.Users
   alias Console.Schema.{
@@ -71,9 +71,11 @@ defmodule Console.Deployments.Cron do
 
   def cache_warm() do
     Task.async(fn -> Git.warm_helm_cache() end)
-    Cluster.healthy()
-    |> Cluster.stream()
-    |> Repo.stream(method: :keyset)
+
+    Cluster.ordered()
+    |> Cluster.healthy()
+    |> Cluster.with_limit(150)
+    |> Repo.all()
     |> Task.async_stream(fn cluster ->
       Logger.info "warming node caches for cluster #{cluster.handle}"
       try do
@@ -86,7 +88,7 @@ defmodule Console.Deployments.Cron do
           Logger.error "hit error trying to warm node caches for cluster=#{cluster.handle}"
           Logger.error(Exception.format(:error, e, __STACKTRACE__))
       end
-    end, [max_concurrency: clamp(Clusters.count())] ++ @opts)
+    end, [max_concurrency: clamp(Clusters.count(), 5, 25)] ++ @opts)
     |> Stream.run()
   end
 

lib/console/deployments/git.ex

@@ -5,6 +5,7 @@ defmodule Console.Deployments.Git do
   alias Console.PubSub
   alias Console.Deployments.{Settings, Services, Clusters}
   alias Console.Services.Users
+  alias Console.Cached.ClusterNodes
   alias Console.Deployments.Pr.{Dispatcher, Validation}
   alias Console.Schema.{
     GitRepository,
@@ -24,7 +25,7 @@ defmodule Console.Deployments.Git do
 
   @cache Console.conf(:cache_adapter)
   @local_cache Console.conf(:local_cache)
-  @ttl :timer.minutes(30)
+  @ttl Console.conf(:ttls)[:helm]
 
   @type repository_resp :: {:ok, GitRepository.t} | Console.error
   @type helm_resp :: {:ok, HelmRepository.t} | Console.error
@@ -395,7 +396,10 @@ defmodule Console.Deployments.Git do
 
   def cached_helm_repositories(), do: @local_cache.get(:helm_repositories)
 
-  def warm_helm_cache(), do: @local_cache.put(:helm_repositories, list_helm_repositories(), ttl: @ttl)
+  def warm_helm_cache() do
+    list_helm_repositories()
+    |> ClusterNodes.helm_repos()
+  end
 
   @spec upsert_helm_repository(binary) :: helm_resp
   def upsert_helm_repository(url) do

lib/console/deployments/local/cache.ex

@@ -0,0 +1,62 @@
+defmodule Console.Deployments.Local.Cache do
+  defstruct [:dir, cache: %{}]
+
+  defmodule Line do
+    @expiry [minutes: -10]
+    defstruct [:file, :digest, :touched]
+
+    def new(file, digest) do
+      %__MODULE__{
+        file: file,
+        digest: digest,
+        touched: Timex.now()
+      }
+    end
+
+    def touch(%__MODULE__{} = mod), do: %{mod | touched: Timex.now()}
+
+    def expire(%__MODULE__{file: f}), do: File.rm(f)
+
+    def expired?(%__MODULE__{touched: touched}) do
+      Timex.now()
+      |> Timex.shift(@expiry)
+      |> Timex.after?(touched)
+    end
+  end
+
+  def new() do
+    {:ok, dir} = Briefly.create(directory: true)
+    %__MODULE__{dir: dir, cache: %{}}
+  end
+
+  def fetch(%__MODULE__{cache: lines} = cache, digest, reader) when is_function(reader, 0) do
+    case lines[digest] do
+      %Line{} = l -> {:ok, l, put_in(cache.cache[digest], Line.touch(l))}
+      nil -> write(cache, digest, reader)
+    end
+  end
+
+  def write(%__MODULE__{} = cache, digest, reader) when is_function(reader, 0) do
+    path = Path.join(cache.dir, "#{digest}.tgz")
+    with %File.Stream{} <- safe_copy(path, reader),
+         line <- Line.new(path, digest),
+      do: {:ok, line, put_in(cache.cache[digest], line)}
+  end
+
+  def sweep(%__MODULE__{cache: lines} = cache) do
+    {keep, expire} = Enum.split_with(lines, fn {_, l} -> !Line.expired?(l) end)
+    Enum.each(expire, &Line.expire/1)
+    %{cache | cache: Map.new(keep)}
+  end
+
+  defp safe_copy(path, reader) when is_function(reader, 0) do
+    with {:ok, f} <- reader.() do
+      try do
+        IO.binstream(f, 1024)
+        |> Enum.into(File.stream!(path))
+      after
+        File.close(f)
+      end
+    end
+  end
+end

lib/console/deployments/local/server.ex

@@ -0,0 +1,27 @@
+defmodule Console.Deployments.Local.Server do
+  use GenServer
+  alias Console.Deployments.Local.Cache
+
+  @timeout :timer.seconds(60)
+
+  def start_link(opts \\ []) do
+    GenServer.start(__MODULE__, opts, name: __MODULE__)
+  end
+
+  def init(_) do
+    :timer.send_interval(:timer.minutes(5), :sweep)
+    {:ok, Cache.new()}
+  end
+
+  def fetch(digest, reader) when is_function(reader, 0),
+    do: GenServer.call(__MODULE__, {:fetch, digest, reader}, @timeout)
+
+  def handle_call({:fetch, digest, reader}, _, cache) when is_function(reader, 0) do
+    case Cache.fetch(cache, digest, reader) do
+      {:ok, line, cache} -> {:reply, {:ok, line.file}, cache}
+      err -> {:reply, err, cache}
+    end
+  end
+
+  def handle_info(:sweep, cache), do: {:noreply, Cache.sweep(cache)}
+end

lib/console/deployments/observer/runner.ex

@@ -13,6 +13,7 @@ defmodule Console.Deployments.Observer.Runner do
       finish(observer, val)
     else
       {:poll, {:error, err}} -> add_error(observer, "poll", err)
+      {:poll, :ignore} -> {:ok, observer}
       {:act, {:error, err}} -> add_error(observer, "action", err)
       err ->
         Logger.error "unknown observer error: #{inspect(err)}"

lib/console/graphql/deployments/cluster.ex

@@ -1063,13 +1063,14 @@ defmodule Console.GraphQl.Deployments.Cluster do
     @desc "a relay connection of all clusters visible to the current user"
     connection field :clusters, node_type: :cluster do
       middleware Authenticated
-      arg :q,          :string
-      arg :healthy,    :boolean
-      arg :tag,        :tag_input
-      arg :tag_query,  :tag_query
-      arg :backups,    :boolean
-      arg :project_id, :id
-      arg :parent_id,  :id
+      arg :q,           :string
+      arg :healthy,     :boolean
+      arg :tag,         :tag_input
+      arg :tag_query,   :tag_query
+      arg :backups,     :boolean
+      arg :project_id,  :id
+      arg :parent_id,   :id
+      arg :upgradeable, :boolean
 
       resolve &Deployments.list_clusters/2
     end

lib/console/graphql/deployments/stack.ex

@@ -59,7 +59,8 @@ defmodule Console.GraphQl.Deployments.Stack do
   end
 
   input_object :policy_engine_attributes do
-    field :type, non_null(:policy_engine_type), description: "the policy engine to use with this stack"
+    field :type,         non_null(:policy_engine_type), description: "the policy engine to use with this stack"
+    field :max_severity, :vuln_severity, description: "the maximum allowed severity without failing the stack run"
   end
 
   input_object :stack_run_attributes do
@@ -153,6 +154,7 @@ defmodule Console.GraphQl.Deployments.Stack do
     field :resource, non_null(:string)
     field :start,    non_null(:integer)
     field :end,      non_null(:integer)
+    field :filename, :string
 
     field :lines, list_of(:stack_violation_cause_line_attributes)
   end
@@ -450,6 +452,7 @@ defmodule Console.GraphQl.Deployments.Stack do
     field :resource, non_null(:string)
     field :start,    non_null(:integer)
     field :end,      non_null(:integer)
+    field :filename, :string
 
     field :lines, list_of(:stack_violation_cause_line)
   end

lib/console/graphql/resolvers/deployments.ex

@@ -73,7 +73,8 @@ defmodule Console.GraphQl.Resolvers.Deployments do
     CloudAddon,
     ClusterScalingRecommendation,
     StackPolicyViolation,
-    StackViolationCause
+    StackViolationCause,
+    Alert
   }
 
   def query(Project, _), do: Project
@@ -145,6 +146,7 @@ defmodule Console.GraphQl.Resolvers.Deployments do
   def query(ClusterScalingRecommendation, _), do: ClusterScalingRecommendation
   def query(StackPolicyViolation, _), do: StackPolicyViolation
   def query(StackViolationCause, _), do: StackViolationCause
+  def query(Alert, _), do: Alert
   def query(_, _), do: Cluster
 
   delegates Console.GraphQl.Resolvers.Deployments.Git