Implement pluggable log aggregation drivers

Starting w/ victoria metrics, and will also implement elastic.  Support authz against main plural objects, eg services, clusters, projects,
 and query constructs that would work with a fully fledged log agg ui.

assets/src/generated/graphql.ts

@@ -2351,6 +2351,8 @@ export type DeploymentSettings = {
   insertedAt?: Maybe<Scalars['DateTime']['output']>;
   /** the latest known k8s version */
   latestK8sVsn: Scalars['String']['output'];
+  /** settings for connections to log aggregation datastores */
+  logging?: Maybe<LoggingSettings>;
   /** the way we can connect to your loki instance */
   lokiConnection?: Maybe<HttpConnection>;
   name: Scalars['String']['output'];
@@ -3305,6 +3307,17 @@ export type LoadBalancerStatus = {
   ingress?: Maybe<Array<Maybe<LoadBalancerIngressStatus>>>;
 };
 
+export enum LogDriver {
+  Elastic = 'ELASTIC',
+  Victoria = 'VICTORIA'
+}
+
+export type LogFacet = {
+  __typename?: 'LogFacet';
+  key: Scalars['String']['output'];
+  value: Scalars['String']['output'];
+};
+
 export type LogFilter = {
   __typename?: 'LogFilter';
   metadata: Metadata;
@@ -3325,12 +3338,36 @@ export type LogLabel = {
   value?: Maybe<Scalars['String']['output']>;
 };
 
+export type LogLine = {
+  __typename?: 'LogLine';
+  facets?: Maybe<Array<Maybe<LogFacet>>>;
+  log: Scalars['String']['output'];
+  timestamp?: Maybe<Scalars['DateTime']['output']>;
+};
+
 export type LogStream = {
   __typename?: 'LogStream';
   stream?: Maybe<Scalars['Map']['output']>;
   values?: Maybe<Array<Maybe<MetricResult>>>;
 };
 
+export type LogTimeRange = {
+  after?: InputMaybe<Scalars['String']['input']>;
+  before?: InputMaybe<Scalars['String']['input']>;
+  duration?: InputMaybe<Scalars['String']['input']>;
+  reverse?: InputMaybe<Scalars['Boolean']['input']>;
+};
+
+/** Settings for configuring log aggregation throughout Plural */
+export type LoggingSettings = {
+  __typename?: 'LoggingSettings';
+  /** the type of log aggregation solution you wish to use */
+  driver?: Maybe<LogDriver>;
+  enabled?: Maybe<Scalars['Boolean']['output']>;
+  /** configures a connection to victoria metrics */
+  victoria?: Maybe<HttpConnection>;
+};
+
 export type LoginInfo = {
   __typename?: 'LoginInfo';
   external?: Maybe<Scalars['Boolean']['output']>;
@@ -6802,6 +6839,7 @@ export type RootQueryType = {
   installations?: Maybe<InstallationConnection>;
   invite?: Maybe<Invite>;
   job?: Maybe<Job>;
+  logAggregation?: Maybe<Array<Maybe<LogLine>>>;
   logFilters?: Maybe<Array<Maybe<LogFilter>>>;
   loginInfo?: Maybe<LoginInfo>;
   logs?: Maybe<Array<Maybe<LogStream>>>;
@@ -7358,6 +7396,15 @@ export type RootQueryTypeJobArgs = {
 };
 
 
+export type RootQueryTypeLogAggregationArgs = {
+  clusterId?: InputMaybe<Scalars['ID']['input']>;
+  limit?: InputMaybe<Scalars['Int']['input']>;
+  query?: InputMaybe<Scalars['String']['input']>;
+  serviceId?: InputMaybe<Scalars['ID']['input']>;
+  time?: InputMaybe<LogTimeRange>;
+};
+
+
 export type RootQueryTypeLogFiltersArgs = {
   namespace: Scalars['String']['input'];
 };

lib/console/graphql/deployments/settings.ex

@@ -2,8 +2,10 @@ defmodule Console.GraphQl.Deployments.Settings do
   use Console.GraphQl.Schema.Base
   alias Console.Deployments.Settings
   alias Console.GraphQl.Resolvers.{Deployments}
+  alias Console.Schema.DeploymentSettings
 
-  ecto_enum :ai_provider, Console.Schema.DeploymentSettings.AIProvider
+  ecto_enum :ai_provider, DeploymentSettings.AIProvider
+  ecto_enum :log_driver, DeploymentSettings.LogDriver
 
   input_object :project_attributes do
     field :name,          non_null(:string)
@@ -49,6 +51,12 @@ defmodule Console.GraphQl.Deployments.Settings do
     field :recommendation_cushion,   :integer, description: "the percentage change needed to generate a recommendation, default 20%"
   end
 
+  input_object :logging_settings_attributes do
+    field :enabled,  :boolean
+    field :driver,   :log_driver
+    field :victoria, :http_connection_attributes
+  end
+
   input_object :ai_settings_attributes do
     field :enabled,       :boolean
     field :tools,         :tool_config_attributes
@@ -154,6 +162,7 @@ defmodule Console.GraphQl.Deployments.Settings do
     field :smtp,                  :smtp_settings, description: "smtp server configuration for email notifications"
     field :ai,                    :ai_settings, description: "settings for LLM provider clients"
     field :cost,                  :cost_settings, description: "settings for cost management"
+    field :logging,               :logging_settings, description: "settings for connections to log aggregation datastores"
 
     field :agent_vsn, non_null(:string), description: "The console's expected agent version",
       resolve: fn _, _, _ -> {:ok, Settings.agent_vsn()} end
@@ -259,6 +268,13 @@ defmodule Console.GraphQl.Deployments.Settings do
     field :location,   non_null(:string), description: "the gcp region the model"
   end
 
+  @desc "Settings for configuring log aggregation throughout Plural"
+  object :logging_settings do
+    field :enabled,  :boolean
+    field :driver,   :log_driver, description: "the type of log aggregation solution you wish to use"
+    field :victoria, :http_connection, description: "configures a connection to victoria metrics"
+  end
+
   connection node_type: :project
 
   object :settings_queries do

lib/console/graphql/observability.ex

@@ -13,6 +13,13 @@ defmodule Console.GraphQl.Observability do
     field :value, :string
   end
 
+  input_object :log_time_range do
+    field :before,   :string
+    field :after,    :string
+    field :duration, :string
+    field :reverse,  :boolean
+  end
+
   object :dashboard do
     field :id,   non_null(:string), resolve: fn %{metadata: %{name: n}}, _, _ -> {:ok, n} end
     field :spec, non_null(:dashboard_spec)
@@ -50,6 +57,17 @@ defmodule Console.GraphQl.Observability do
     end
   end
 
+  object :log_line do
+    field :timestamp, :datetime
+    field :log,       non_null(:string)
+    field :facets,    list_of(:log_facet)
+  end
+
+  object :log_facet do
+    field :key,   non_null(:string)
+    field :value, non_null(:string)
+  end
+
   object :observability_queries do
     field :dashboards, list_of(:dashboard) do
       middleware Authenticated
@@ -83,6 +101,17 @@ defmodule Console.GraphQl.Observability do
       safe_resolve &Observability.resolve_metric/2
     end
 
+    field :log_aggregation, list_of(:log_line) do
+      middleware Authenticated
+      arg :service_id, :id
+      arg :cluster_id, :id
+      arg :query,      :string
+      arg :time,       :log_time_range
+      arg :limit,      :integer
+
+      resolve &Observability.list_logs/2
+    end
+
     field :logs, list_of(:log_stream) do
       middleware Authenticated
 

lib/console/graphql/resolvers/observability.ex

@@ -1,5 +1,7 @@
 defmodule Console.GraphQl.Resolvers.Observability do
   alias Console.Services.Observability
+  alias Console.Logs.{Provider, Query}
+
   @default_offset 30 * 60
   @nano 1_000_000_000
 
@@ -16,6 +18,12 @@ defmodule Console.GraphQl.Resolvers.Observability do
     end
   end
 
+  def list_logs(args, %{context: %{current_user: user}}) do
+    query = Query.new(args)
+    with {:ok, query} <- Provider.accessible(query, user),
+      do: Provider.query(query)
+  end
+
   def resolve_logs(%{query: query, limit: limit} = args, _) do
     now    = Timex.now()
     start  = (args[:start] || ts(now)) / @nano

lib/console/logs/line.ex

@@ -0,0 +1,9 @@
+defmodule Console.Logs.Line do
+  @type t :: %__MODULE__{facets: [%{key: binary, value: binary}]}
+
+  defstruct [:timestamp, :log, :facets]
+
+  def new(map) do
+    %__MODULE__{log: map[:log], timestamp: map[:timestamp], facets: map[:facets]}
+  end
+end

lib/console/logs/provider.ex

@@ -0,0 +1,27 @@
+defmodule Console.Logs.Provider do
+  alias Console.Logs.{Query, Line}
+  alias Console.Logs.Provider.{Victoria}
+  alias Console.Schema.{User, DeploymentSettings}
+
+  @type error :: Console.error
+
+  @callback query(struct, Query.t) :: {:ok, [Line.t]} | error
+
+  @spec query(Query.t) :: {:ok, [Line.t]} | error
+  def query(%Query{} = q) do
+    with {:ok, %{__struct__: provider} = prov} <- client(),
+      do: provider.query(prov, q)
+  end
+
+  @spec accessible(Query.t, User.t) :: {:ok, Query.t} | error
+  def accessible(%Query{} = query, %User{} = user), do: Query.accessible(query, user)
+
+  defp client() do
+    Console.Deployments.Settings.cached()
+    |> client()
+  end
+
+  def client(%DeploymentSettings{logging: %{enabled: true, driver: :victoria, victoria: %{} = victoria}}),
+    do: {:ok, Victoria.new(victoria)}
+  def client(_), do: {:error, "Plural logging integration not yet configured"}
+end

lib/console/logs/provider/utils.ex

@@ -0,0 +1,6 @@
+defmodule Console.Logs.Provider.Utils do
+  def facets(%{} = map) do
+    Enum.map(map, fn {k, v} -> %{key: k, value: v} end)
+  end
+  def facets(_), do: nil
+end