diff --git a/.gitignore b/.gitignore index d734441c96..a401d7ab88 100644 --- a/.gitignore +++ b/.gitignore @@ -63,6 +63,7 @@ gha-creds-*.json !.template.env !.single_tenant_bq.env +!.single_tenant_ch.env !.single_tenant_pg.env # rustler nif binaries /priv/native diff --git a/.reach.baseline.json b/.reach.baseline.json index 01c1591ebe..b2d31227b6 100644 --- a/.reach.baseline.json +++ b/.reach.baseline.json @@ -781,10 +781,10 @@ { "source": "smells", "kind": "redundant_computation", - "fingerprint": "sha256:def6b8334cb5e0de59e58e669a16f009f819b7a3f280e81d65b6cf75216d3646", - "message": "Logflare.Auth.list_valid_access_tokens called twice with same args (line 138 and 168)", + "fingerprint": "sha256:9a7130a7a6e263eec866c2d02c31a837cb2fc4f7ed89ceac5eda162e7d76570e", + "message": "Logflare.Auth.list_valid_access_tokens called twice with same args (line 187 and 217)", "file": "lib/logflare/single_tenant.ex", - "line": 168 + "line": 217 }, { "source": "smells", diff --git a/.single_tenant_ch.env b/.single_tenant_ch.env new file mode 100644 index 0000000000..6bdec1ec33 --- /dev/null +++ b/.single_tenant_ch.env @@ -0,0 +1,19 @@ +PHX_HTTP_PORT=4000 +DB_DATABASE=logflare_dev +DB_HOSTNAME=localhost +DB_PORT=5432 +DB_PASSWORD=postgres +DB_USERNAME=postgres +DB_SCHEMA=public + +LOGFLARE_SINGLE_TENANT=true +LOGFLARE_PRIVATE_ACCESS_TOKEN=my-cool-api-key-123 +LOGFLARE_PUBLIC_ACCESS_TOKEN=my-other-cool-api-key-123 +LOGFLARE_GRPC_PORT=50051 + +CLICKHOUSE_BACKEND_URL=http://logflare:logflare@localhost:8123/logflare_test +CLICKHOUSE_BACKEND_POOL_SIZE=3 + +LOGFLARE_NODE_HOST=127.0.0.1 +LOGFLARE_HTTP_CONNECTION_POOLS=all +LOGFLARE_FEATURE_FLAG_OVERRIDE=multibackend=true,key_values=true diff --git a/DEVELOPMENT.md b/DEVELOPMENT.md index 4e319ad376..1551c7403f 100644 --- a/DEVELOPMENT.md +++ b/DEVELOPMENT.md @@ -53,7 +53,7 @@ docker-compose up -d db clickhouse telegraf # install dependencies make setup -# start in single tenant postgres backend (logflare_dev) +# start in single tenant postgres backend (logflare_dev). See [Developing for Single Tenant](#developing-for-single-tenant) for other backends. make start.st.pg # run tests @@ -73,17 +73,50 @@ To configure the BigQuery backend, please follow the [BigQuery setup documentati ### Developing for Single Tenant -Use the single tenant `make start.*` variations. This works by switching out the `LOGFLARE_SINGLE_TENANT` env var. +Use a backend-specific single tenant target. Each target loads its corresponding +`.single_tenant_*.env` file and enables `LOGFLARE_SINGLE_TENANT`. ```bash make start.st.pg make start.st.bq +make start.st.ch ``` +#### ClickHouse + +Start the application PostgreSQL database and ClickHouse service, then start Logflare: + +```bash +docker compose up -d db clickhouse telegraf +make start.st.ch +``` + +`make start.st.ch` loads `.single_tenant_ch.env`. The main settings are: + +- `CLICKHOUSE_BACKEND_URL` - the ClickHouse HTTP connection URL, including the database name +- `CLICKHOUSE_BACKEND_POOL_SIZE` - optional connection pool size +- `LOGFLARE_CLICKHOUSE_TABLE_SUFFIX` - suffix for the single-tenant `otel_logs`, `otel_metrics`, and `otel_traces` tables (defaults to `default`) +- `LOGFLARE_PUBLIC_ACCESS_TOKEN` and `LOGFLARE_PRIVATE_ACCESS_TOKEN` - local API tokens + +PostgreSQL is still required for Logflare application data. ClickHouse becomes the default +log backend and uses one shared ingestion pipeline configured from `CLICKHOUSE_BACKEND_URL`; the +default ClickHouse backend is not stored as a database record. + +The local URL is typically: + +```text +http://logflare:logflare@localhost:8123/logflare_test +``` + +If both `POSTGRES_BACKEND_URL` and `CLICKHOUSE_BACKEND_URL` are present in the effective +environment, PostgreSQL takes precedence. Unset `POSTGRES_BACKEND_URL` when using +`make start.st.ch`. + To develop with Supabase mode: ```bash make start.sb.bq +make start.sb.ch make start.sb.pg ``` @@ -164,7 +197,8 @@ bin/compose down -v make setup make start make start.{orange|pink} -make start.{st|sb}.{bq|pg} +make start.st.{bq|ch|pg} +make start.sb.{bq|ch|pg} make decrypt.{dev|staging|prod} make encrypt.{dev|staging|prod} make reset diff --git a/Makefile b/Makefile index 8c39f67cb3..ccd67dc298 100644 --- a/Makefile +++ b/Makefile @@ -120,7 +120,7 @@ setup: check-tools check-version-manager setup.node setup.node: @echo -e "$(BOLD)$(BLUE)📦 Installing Node.js dependencies...$(NC)" @echo "" - npm --prefix ./assets ci + npm --prefix ./assets ci @echo "" reset: @@ -157,6 +157,9 @@ start.green: __start__ start.sb.bq: LOGFLARE_SUPABASE_MODE = true start.sb.bq: start.st.bq +start.sb.ch: LOGFLARE_SUPABASE_MODE = true +start.sb.ch: start.st.ch + # SPOOL_MODE (producer|consumer|both), SPOOL_PROVIDER (aws|gcp), and related # SPOOL_* vars pass straight through to the app via __start__ (see config/runtime.exs). # To run a producer + consumer pair side by side, override PORT/ERL_NAME/LOGFLARE_GRPC_PORT @@ -169,6 +172,12 @@ start.st.bq: ENV_FILE = .single_tenant_bq.env start.st.bq: LOGFLARE_GRPC_PORT ?= 50051 start.st.bq: __start__ +start.st.ch: ERL_NAME ?= st_ +start.st.ch: PORT ?= 4000 +start.st.ch: ENV_FILE = .single_tenant_ch.env +start.st.ch: LOGFLARE_GRPC_PORT ?= 50051 +start.st.ch: __start__ + start.sb.pg: LOGFLARE_SUPABASE_MODE = true start.sb.pg: start.st.pg @@ -195,7 +204,7 @@ migrate: stripe: stripe listen --forward-to localhost:4000/webhooks/stripe -.PHONY: __start__ migrate stripe start.sb.pg start.sb.bq start.st.pg start.st.bq start.orange start.pink +.PHONY: __start__ migrate stripe start.sb.pg start.sb.ch start.sb.bq start.st.pg start.st.ch start.st.bq start.orange start.pink # Encryption and decryption of secrets # Usage: @@ -463,12 +472,12 @@ $(TELEGRAF_KEYS_SENTINEL): # Server (Telegraf) @echo "subjectAltName=DNS:localhost,IP:127.0.0.1,DNS:telegraf" > $(TELEGRAF_KEYS_DIR)/server.ext - + @openssl genrsa -out $(TELEGRAF_KEYS_DIR)/server.key 2048 @openssl req -new -key $(TELEGRAF_KEYS_DIR)/server.key \ -subj "/CN=localhost" \ -out $(TELEGRAF_KEYS_DIR)/server.csr - + @openssl x509 -req -days 3650 -in $(TELEGRAF_KEYS_DIR)/server.csr \ -CA $(TELEGRAF_KEYS_DIR)/ca.crt -CAkey $(TELEGRAF_KEYS_DIR)/ca.key -set_serial 01 \ -out $(TELEGRAF_KEYS_DIR)/server.crt \ diff --git a/config/runtime.exs b/config/runtime.exs index aae7896e1c..dc4488f586 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -1,5 +1,6 @@ import Config +alias Logflare.SingleTenant alias Logflare.Utils defmodule Env do @@ -135,6 +136,7 @@ config :logflare, recaptcha_secret: System.get_env("LOGFLARE_RECAPTCHA_SECRET"), config_cat_sdk_key: System.get_env("LOGFLARE_CONFIG_CAT_SDK_KEY"), single_tenant: Env.get_boolean("LOGFLARE_SINGLE_TENANT"), + clickhouse_table_suffix: System.get_env("LOGFLARE_CLICKHOUSE_TABLE_SUFFIX", "default"), supabase_mode: Env.get_boolean("LOGFLARE_SUPABASE_MODE"), unsafe_disable_ssrf_s3_endpoint_check: Env.get_boolean("LOGFLARE_UNSAFE_DISABLE_SSRF_S3_ENDPOINT_CHECK"), @@ -390,15 +392,54 @@ socket_options_for_url = fn nil end +single_tenant? = Env.get_boolean("LOGFLARE_SINGLE_TENANT") +bigquery_service_account = System.get_env("GOOGLE_SERVICE_ACCOUNT") +postgres_backend_url = System.get_env("POSTGRES_BACKEND_URL") +clickhouse_backend_url = System.get_env("CLICKHOUSE_BACKEND_URL") + +backend_url_set? = fn url -> is_binary(url) and String.trim(url) != "" end + +if Enum.count( + [bigquery_service_account, postgres_backend_url, clickhouse_backend_url], + backend_url_set? + ) > 1 do + raise "Only one of GOOGLE_SERVICE_ACCOUNT, POSTGRES_BACKEND_URL, or CLICKHOUSE_BACKEND_URL may be configured" +end + +single_tenant_backend = + cond do + single_tenant? and backend_url_set?.(postgres_backend_url) -> :postgres + single_tenant? and backend_url_set?.(clickhouse_backend_url) -> :clickhouse + true -> :bigquery + end + +config :logflare, :single_tenant_backend, single_tenant_backend + cond do - Env.get_boolean("LOGFLARE_SINGLE_TENANT") && - not is_nil(System.get_env("POSTGRES_BACKEND_URL")) -> + single_tenant? and single_tenant_backend == :postgres -> config :logflare, :postgres_backend_adapter, filter_nil_kv_pairs.( - url: System.get_env("POSTGRES_BACKEND_URL"), + url: postgres_backend_url, schema: System.get_env("POSTGRES_BACKEND_SCHEMA"), - socket_options: socket_options_for_url.(System.get_env("POSTGRES_BACKEND_URL", "")) + socket_options: socket_options_for_url.(postgres_backend_url) + ) + + single_tenant? and single_tenant_backend == :clickhouse -> + clickhouse_backend_config = + SingleTenant.clickhouse_backend_adapter_opts_from_url!(clickhouse_backend_url) + + config :logflare, + :clickhouse_backend_adapter, + filter_nil_kv_pairs.( + clickhouse_backend_config ++ + [ + pool_size: + case System.get_env("CLICKHOUSE_BACKEND_POOL_SIZE") do + nil -> nil + value -> String.to_integer(value) + end + ] ) config_env() != :test -> diff --git a/lib/logflare/application.ex b/lib/logflare/application.ex index 51ced4df32..7ee7c8fa83 100644 --- a/lib/logflare/application.ex +++ b/lib/logflare/application.ex @@ -9,6 +9,7 @@ defmodule Logflare.Application do alias Logflare.Backends.Adaptor.BigQueryAdaptor alias Logflare.Backends.Spool.Health, as: SpoolHealth alias Logflare.Backends.Spool.SpoolAck + alias Logflare.Backends.SystemBackend alias Logflare.Backends.UserMonitoring alias Logflare.ContextCache alias Logflare.Logs @@ -234,15 +235,13 @@ defmodule Logflare.Application do def startup_tasks do Logger.info("Executing startup tasks") - if !SingleTenant.postgres_backend?() do - BigQueryAdaptor.create_managed_service_accounts() - BigQueryAdaptor.update_iam_policy() - end + SystemBackend.on_system_start() if SingleTenant.single_tenant?() do Logger.info("Ensuring single tenant user is seeded...") SingleTenant.create_default_plan() SingleTenant.create_default_user() + SingleTenant.ensure_default_backend() SingleTenant.create_access_tokens() end @@ -251,13 +250,7 @@ defmodule Logflare.Application do SingleTenant.create_supabase_endpoints() SingleTenant.ensure_supabase_sources_started() - unless SingleTenant.postgres_backend?() do - # buffer time for all sources to init and create tables - # in case of latency. - :timer.sleep(3_000) - - SingleTenant.update_supabase_source_schemas() - end + SystemBackend.on_supabase_start() end # Schedule all alerts immediately on startup diff --git a/lib/logflare/backends.ex b/lib/logflare/backends.ex index e94f1cbcd7..88565704f4 100644 --- a/lib/logflare/backends.ex +++ b/lib/logflare/backends.ex @@ -44,6 +44,9 @@ defmodule Logflare.Backends do @max_pipeline_swap_delay_ms 5_000 @type one_or_list_or_nil :: Backend.t() | [Backend.t()] | nil + @type ingest_queue_key :: + IngestEventQueue.queues_key() | IngestEventQueue.consolidated_queues_key() + @type queue_owner :: non_neg_integer() | :consolidated @doc """ Retrieves the hardcoded max pending buffer length of an individual queue @@ -51,6 +54,39 @@ defmodule Logflare.Backends do @spec max_buffer_queue_len() :: non_neg_integer() def max_buffer_queue_len, do: @max_pending_buffer_len_per_queue + @doc "Returns whether BigQuery is the deployment's default backend" + @spec bigquery_default_backend?() :: boolean() + def bigquery_default_backend?, + do: not SingleTenant.single_tenant?() or SingleTenant.bigquery_backend?() + + @doc """ + Returns queue key used to buffer events for a source and backend. + + Consolidated backends share a queue keyed by backend ID, other backends scope + to source and backend. + """ + @spec ingest_queue_key(Source.t(), Backend.t() | nil) :: ingest_queue_key() + def ingest_queue_key(%Source{}, %Backend{consolidated_ingest?: true, id: backend_id}) do + {:consolidated, backend_id} + end + + def ingest_queue_key(%Source{id: source_id}, %Backend{id: backend_id}) do + {source_id, backend_id} + end + + def ingest_queue_key(%Source{id: source_id}, nil), do: {source_id, nil} + + @doc """ + Returns the queue key used to buffer events for a source's system default backend. + + In single-tenant ClickHouse deployments, uses `ingest_queue_key/2` with the + source user's default backend. Otherwise, returns `{source_id, nil}`. + """ + @spec system_default_ingest_queue_key(Source.t()) :: ingest_queue_key() + def system_default_ingest_queue_key(%Source{} = source) do + ingest_queue_key(source, SingleTenant.lookup_system_default_backend(source)) + end + @doc """ Returns the maximum time, in microseconds, that an event's timestamp may be in the future and still be accepted at ingestion. Events further ahead than this @@ -244,29 +280,32 @@ defmodule Logflare.Backends do end end + @spec get_default_backend(User.t()) :: Backend.t() | nil def get_default_backend(%User{} = user) do - if SingleTenant.single_tenant?() and SingleTenant.postgres_backend?() do - opts = SingleTenant.postgres_backend_adapter_opts() - - %Backend{ - type: :postgres, - config: Map.new(opts), - user_id: user.id, - name: "Default PostgreSQL backend" - } - else - project_id = user.bigquery_project_id || User.bq_project_id() - dataset_id = user.bigquery_dataset_id || User.generate_bq_dataset_id(user.id) - - %Backend{ - type: :bigquery, - config: %{ - project_id: project_id, - dataset_id: dataset_id - }, - user_id: user.id, - name: "Default BigQuery backend" - } + cond do + SingleTenant.clickhouse_backend?() -> + SingleTenant.get_default_clickhouse_backend(user) + + SingleTenant.postgres_backend?() -> + opts = SingleTenant.postgres_backend_adapter_opts() + + %Backend{ + type: :postgres, + config: Map.new(opts), + user_id: user.id, + name: "Default PostgreSQL backend" + } + + true -> + project_id = user.bigquery_project_id || User.bq_project_id() + dataset_id = user.bigquery_dataset_id || User.generate_bq_dataset_id(user.id) + + %Backend{ + type: :bigquery, + config: %{project_id: project_id, dataset_id: dataset_id}, + user_id: user.id, + name: "Default BigQuery backend" + } end end @@ -536,6 +575,8 @@ defmodule Logflare.Backends do Retrieves a Backend by id. """ @spec get_backend(integer()) :: Backend.t() | nil + def get_backend(0), do: SingleTenant.get_default_clickhouse_backend() + def get_backend(id) do backend = Repo.get(Backend, id) @@ -992,7 +1033,7 @@ defmodule Logflare.Backends do defp dispatch_to_backends(source, nil, log_events) do backends = __MODULE__.Cache.list_backends(source_id: source.id) - for backend <- [nil | backends] do + for backend <- [SingleTenant.lookup_system_default_backend(source) | backends] do dispatch_to_default_backend(source, backend, log_events) end end @@ -1003,17 +1044,8 @@ defmodule Logflare.Backends do log_events :: [LogEvent.t()] ) :: any() defp dispatch_to_default_backend(source, backend, log_events) do - {queue_key, backend_type} = - case backend do - nil -> - {{source.id, nil}, SingleTenant.backend_type()} - - %Backend{consolidated_ingest?: true} -> - {{:consolidated, backend.id}, backend.type} - - %Backend{} -> - {{source.id, backend.id}, backend.type} - end + queue_key = ingest_queue_key(source, backend) + backend_type = if backend, do: backend.type, else: SingleTenant.backend_type() telemetry_metadata = %{backend_type: backend_type} @@ -1088,7 +1120,7 @@ defmodule Logflare.Backends do def via_backend(backend_id, mod, nil), do: via_backend(backend_id, mod) - def via_backend(backend_id, mod, label) when is_pos_integer(backend_id) do + def via_backend(backend_id, mod, label) when is_non_negative_integer(backend_id) do {:via, Registry, {BackendRegistry, {mod, backend_id, label}}} end @@ -1404,35 +1436,32 @@ defmodule Logflare.Backends do - Returns true only if ALL queues are full """ @spec cached_local_pending_buffer_full?(Source.t()) :: boolean() - def cached_local_pending_buffer_full?(%Source{ - id: source_id, - default_ingest_backend_enabled?: true - }) do - default_backend_ids = - __MODULE__.Cache.list_backends(source_id: source_id) + def cached_local_pending_buffer_full?( + %Source{ + default_ingest_backend_enabled?: true + } = source + ) do + default_backend_queue_keys = + __MODULE__.Cache.list_backends(source_id: source.id) |> Enum.filter(& &1.default_ingest?) - |> MapSet.new(& &1.id) + |> MapSet.new(&ingest_queue_key(source, &1)) - # Check system default backend (nil backend_id) - system_default_full? = buffer_full_for_backend?(source_id, nil) + system_default_full? = buffer_full_for_queue?(system_default_ingest_queue_key(source)) - # Check user-configured default backends - user_defaults_full? = Enum.any?(default_backend_ids, &buffer_full_for_backend?(source_id, &1)) + user_defaults_full? = Enum.any?(default_backend_queue_keys, &buffer_full_for_queue?/1) system_default_full? || user_defaults_full? end - def cached_local_pending_buffer_full?(%Source{id: source_id}) do - buffer_full_for_backend?(source_id, nil) + def cached_local_pending_buffer_full?(%Source{} = source) do + source + |> system_default_ingest_queue_key() + |> buffer_full_for_queue?() end - @spec buffer_full_for_backend?( - source_id :: non_neg_integer(), - backend_id :: non_neg_integer() | nil - ) :: - boolean() - defp buffer_full_for_backend?(source_id, backend_id) do - case PubSubRates.Cache.get_local_buffer(source_id, backend_id) do + @spec buffer_full_for_queue?(ingest_queue_key()) :: boolean() + defp buffer_full_for_queue?({queue_owner, backend_id}) do + case PubSubRates.Cache.get_local_buffer(queue_owner, backend_id) do %{queues: [_ | _] = queues} -> Enum.all?(queues, fn {_key, count} -> count > @max_pending_buffer_len_per_queue @@ -1456,14 +1485,26 @@ defmodule Logflare.Backends do """ @spec any_ingest_queue_over_limit?(pos_integer()) :: boolean() def any_ingest_queue_over_limit?(source_id) do - backend_ids = - [nil | __MODULE__.Cache.list_backends(source_id: source_id) |> Enum.map(& &1.id)] + source = Sources.Cache.get_by_id(source_id) + + queue_keys = + case source do + %Source{} -> + backend_queue_keys = + __MODULE__.Cache.list_backends(source_id: source_id) + |> Enum.map(&ingest_queue_key(source, &1)) - Enum.any?(backend_ids, &any_queue_over_limit_for_backend?(source_id, &1)) + [system_default_ingest_queue_key(source) | backend_queue_keys] + + nil -> + if SingleTenant.clickhouse_backend?(), do: [], else: [{source_id, nil}] + end + + Enum.any?(queue_keys, &any_queue_over_limit?/1) end - defp any_queue_over_limit_for_backend?(source_id, backend_id) do - {source_id, backend_id} + defp any_queue_over_limit?(queue_key) do + queue_key |> IngestEventQueue.list_queues() |> Enum.any?(&queue_over_limit?/1) end @@ -1478,29 +1519,30 @@ defmodule Logflare.Backends do @doc """ Caches total buffer len. Includes ingested events that are awaiting cleanup. """ - @spec cache_local_buffer_lens(non_neg_integer(), non_neg_integer() | nil) :: + @spec cache_local_buffer_lens(queue_owner(), non_neg_integer() | nil) :: {:ok, %{ len: non_neg_integer(), - queues: [{Logflare.Backends.IngestEventQueue.table_key(), non_neg_integer()}] + queues: [{tuple(), non_neg_integer()}] }} - def cache_local_buffer_lens(source_id, backend_id \\ nil) do - queues = IngestEventQueue.list_counts({source_id, backend_id}) + def cache_local_buffer_lens(queue_owner, backend_id \\ nil) do + queues = IngestEventQueue.list_counts({queue_owner, backend_id}) len = for({_k, v} <- queues, do: v) |> Enum.sum() stats = %{len: len, queues: queues} payload = %{Node.self() => stats} - PubSubRates.Cache.cache_buffers(source_id, backend_id, payload) + PubSubRates.Cache.cache_buffers(queue_owner, backend_id, payload) {:ok, stats} end @doc """ Get local pending buffer len of a source/backend combination. """ - @spec cached_local_pending_buffer_len(Source.t(), Backend.t() | nil) :: map() - def cached_local_pending_buffer_len(source_id, backend_id \\ nil) when is_integer(source_id) do - PubSubRates.Cache.get_local_buffer(source_id, backend_id) + @spec cached_local_pending_buffer_len(queue_owner(), non_neg_integer() | nil) :: map() + def cached_local_pending_buffer_len(queue_owner, backend_id \\ nil) + when is_integer(queue_owner) or queue_owner == :consolidated do + PubSubRates.Cache.get_local_buffer(queue_owner, backend_id) end @doc """ @@ -1508,7 +1550,12 @@ defmodule Logflare.Backends do """ @spec cached_pending_buffer_len(Source.t(), Backend.t() | nil) :: non_neg_integer() def cached_pending_buffer_len(%Source{} = source, backend \\ nil) do - PubSubRates.Cache.get_cluster_buffers(source.id, Map.get(backend || %{}, :id)) + {queue_owner, backend_id} = + if backend, + do: ingest_queue_key(source, backend), + else: system_default_ingest_queue_key(source) + + PubSubRates.Cache.get_cluster_buffers(queue_owner, backend_id) end @doc """ diff --git a/lib/logflare/backends/adaptor/bigquery_adaptor.ex b/lib/logflare/backends/adaptor/bigquery_adaptor.ex index 929effdd83..9d9bf26f0a 100644 --- a/lib/logflare/backends/adaptor/bigquery_adaptor.ex +++ b/lib/logflare/backends/adaptor/bigquery_adaptor.ex @@ -2,6 +2,7 @@ defmodule Logflare.Backends.Adaptor.BigQueryAdaptor do @moduledoc false @behaviour Logflare.Backends.Adaptor + @behaviour Logflare.Backends.SystemBackend use Supervisor @@ -28,15 +29,19 @@ defmodule Logflare.Backends.Adaptor.BigQueryAdaptor do alias Logflare.BqRepo alias Logflare.Endpoints.EndpointQuery alias Logflare.Google + alias Logflare.Google.BigQuery alias Logflare.Google.BigQuery.EventUtils alias Logflare.Google.BigQuery.GCPConfig alias Logflare.Google.BigQuery.GenUtils alias Logflare.Google.CloudResourceManager + alias Logflare.SingleTenant alias Logflare.Sources + alias Logflare.Sources.Source alias Logflare.Sources.Source.BigQuery.Pipeline alias Logflare.Sources.Source.BigQuery.Schema alias Logflare.User alias Logflare.Users + alias Logflare.Utils.Tasks alias Model.QueryParameter, as: Param alias Model.QueryParameterType, as: Type alias Model.QueryParameterValue, as: Value @@ -47,6 +52,47 @@ defmodule Logflare.Backends.Adaptor.BigQueryAdaptor do @search_query_timeout_ms 60_000 @endpoint_query_timeout_ms 60_000 + @impl Logflare.Backends.SystemBackend + @spec on_system_start() :: :ok + def on_system_start do + create_managed_service_accounts() + update_iam_policy() + :ok + end + + @impl Logflare.Backends.SystemBackend + @spec on_source_start(Source.t()) :: :ok + def on_source_start(%Source{token: source_token}) do + %{ + user_id: user_id, + bigquery_table_ttl: bigquery_table_ttl, + bigquery_project_id: bigquery_project_id, + bigquery_dataset_location: bigquery_dataset_location, + bigquery_dataset_id: bigquery_dataset_id + } = GenUtils.get_bq_user_info(source_token) + + Tasks.start_child(fn -> + BigQuery.init_table!( + user_id, + source_token, + bigquery_project_id, + bigquery_table_ttl, + bigquery_dataset_location, + bigquery_dataset_id + ) + end) + + :ok + end + + @impl Logflare.Backends.SystemBackend + @spec on_supabase_start() :: :ok + def on_supabase_start do + :timer.sleep(3_000) + SingleTenant.update_supabase_source_schemas() + :ok + end + @impl Logflare.Backends.Adaptor def start_link({source, backend} = source_backend) do Supervisor.start_link(__MODULE__, source_backend, diff --git a/lib/logflare/backends/adaptor/clickhouse_adaptor.ex b/lib/logflare/backends/adaptor/clickhouse_adaptor.ex index 44a51449cc..2d73f250e2 100644 --- a/lib/logflare/backends/adaptor/clickhouse_adaptor.ex +++ b/lib/logflare/backends/adaptor/clickhouse_adaptor.ex @@ -36,6 +36,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor do alias Logflare.Backends.QueryError alias Logflare.LogEvent alias Logflare.LogEvent.TypeDetection + alias Logflare.SingleTenant alias Logflare.Sources.Source alias Logflare.Sql.DialectTransformer.ClickHouse, as: ClickHouseSqlTransformer alias Mint.Types, as: MintTypes @@ -706,16 +707,23 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor do - `:log` -> `otel_logs_` - `:metric` -> `otel_metrics_` - `:trace` -> `otel_traces_` + + Single-tenant defaults use the configured table suffix in place of `_`. """ @spec clickhouse_ingest_table_name(Backend.t(), TypeDetection.event_type()) :: String.t() - def clickhouse_ingest_table_name(%Backend{} = backend, :log), - do: build_otel_table_name(backend, "otel_logs") + def clickhouse_ingest_table_name(%Backend{} = backend, event_type) + when event_type in [:log, :metric, :trace] do + prefix = "otel_" <> to_string(event_type) <> "s" - def clickhouse_ingest_table_name(%Backend{} = backend, :metric), - do: build_otel_table_name(backend, "otel_metrics") + if SingleTenant.default_clickhouse_backend?(backend) do + prefix <> "_" <> clickhouse_table_suffix() + else + build_otel_table_name(backend, prefix) + end + end - def clickhouse_ingest_table_name(%Backend{} = backend, :trace), - do: build_otel_table_name(backend, "otel_traces") + @spec clickhouse_table_suffix() :: String.t() + defp clickhouse_table_suffix, do: Application.fetch_env!(:logflare, :clickhouse_table_suffix) @spec build_otel_table_name(Backend.t(), String.t()) :: String.t() defp build_otel_table_name(%Backend{token: token}, prefix) do @@ -1608,10 +1616,10 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor do end end - @spec maybe_start_query_connection_manager(pid() | nil, pos_integer(), String.t() | nil) :: + @spec maybe_start_query_connection_manager(pid() | nil, non_neg_integer(), String.t() | nil) :: :ok | {:error, term()} defp maybe_start_query_connection_manager(nil, backend_id, label) - when is_pos_integer(backend_id) do + when is_non_negative_integer(backend_id) do backend = Backends.Cache.get_backend(backend_id) with child_spec <- ConnectionManager.child_spec(backend, label), @@ -1639,7 +1647,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor do end end - defp maybe_start_query_connection_manager(_pid, _backend_id, _label), do: :ok + defp maybe_start_query_connection_manager(pid, _backend_id, _label) when is_pid(pid), do: :ok @spec ensure_pool_and_notify(Backend.t(), String.t() | nil) :: :ok | {:error, term()} defp ensure_pool_and_notify(%Backend{} = backend, label) do diff --git a/lib/logflare/backends/adaptor/clickhouse_adaptor/circuit_breaker.ex b/lib/logflare/backends/adaptor/clickhouse_adaptor/circuit_breaker.ex index 9fbeaf8df7..b04c64a7bc 100644 --- a/lib/logflare/backends/adaptor/clickhouse_adaptor/circuit_breaker.ex +++ b/lib/logflare/backends/adaptor/clickhouse_adaptor/circuit_breaker.ex @@ -26,7 +26,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.CircuitBreaker do @block_ms :timer.seconds(30) typedstruct do - field :backend_id, pos_integer(), enforce: true + field :backend_id, non_neg_integer(), enforce: true field :table, :ets.table(), enforce: true field :failures, [integer()], default: [] end @@ -52,10 +52,10 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.CircuitBreaker do Reads the breaker's ETS table directly via the backend registry, so a missing or crashed breaker reads as allowed. """ - @spec check(Backend.t() | pos_integer()) :: :ok | {:error, :circuit_open, integer()} + @spec check(Backend.t() | non_neg_integer()) :: :ok | {:error, :circuit_open, integer()} def check(%Backend{id: backend_id}), do: check(backend_id) - def check(backend_id) when is_pos_integer(backend_id) do + def check(backend_id) when is_non_negative_integer(backend_id) do with [{_pid, table}] when not is_nil(table) <- Registry.lookup(BackendRegistry, {__MODULE__, backend_id}), [{@blocked_key, blocked_until}] <- safe_lookup(table, @blocked_key), @@ -112,10 +112,10 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.CircuitBreaker do Useful for introspection (_and tests_). """ - @spec get_state(Backend.t() | pos_integer()) :: t() | nil + @spec get_state(Backend.t() | non_neg_integer()) :: t() | nil def get_state(%Backend{id: backend_id}), do: get_state(backend_id) - def get_state(backend_id) when is_pos_integer(backend_id) do + def get_state(backend_id) when is_non_negative_integer(backend_id) do case Registry.lookup(BackendRegistry, {__MODULE__, backend_id}) do [{pid, _table}] -> GenServer.call(pid, :get_state) [] -> nil diff --git a/lib/logflare/backends/adaptor/clickhouse_adaptor/connection_manager.ex b/lib/logflare/backends/adaptor/clickhouse_adaptor/connection_manager.ex index 0676b52eb5..d37bf86299 100644 --- a/lib/logflare/backends/adaptor/clickhouse_adaptor/connection_manager.ex +++ b/lib/logflare/backends/adaptor/clickhouse_adaptor/connection_manager.ex @@ -46,7 +46,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.ConnectionManager do @telemetry_listener __MODULE__.TelemetryListener typedstruct do - field :backend_id, pos_integer(), enforce: true + field :backend_id, non_neg_integer(), enforce: true field :label, String.t() | nil, default: nil field :pool_pid, pid() | nil, default: nil field :last_activity, integer() | nil, default: nil @@ -573,7 +573,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.ConnectionManager do urls = Map.get(config, :read_only_urls) || %{} default = Map.get(config, :default_read_cluster) - labeled_url(urls, label) || labeled_url(urls, default) || Map.get(config, :url) + labeled_url(urls, label) || labeled_url(urls, default) || legacy_read_url(config) end @spec labeled_url(map(), String.t() | nil) :: String.t() | nil @@ -586,6 +586,12 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.ConnectionManager do defp labeled_url(_urls, _label), do: nil + @spec legacy_read_url(map()) :: String.t() | nil + defp legacy_read_url(config) do + read_only_url = Map.get(config, :read_only_url) + if is_non_empty_binary(read_only_url), do: read_only_url, else: Map.get(config, :url) + end + @spec connection_host(pos_integer(), String.t() | nil) :: String.t() | nil defp connection_host(backend_id, label) do backend_id diff --git a/lib/logflare/backends/adaptor/clickhouse_adaptor/pipeline.ex b/lib/logflare/backends/adaptor/clickhouse_adaptor/pipeline.ex index 51a7005a79..a7a093cf47 100644 --- a/lib/logflare/backends/adaptor/clickhouse_adaptor/pipeline.ex +++ b/lib/logflare/backends/adaptor/clickhouse_adaptor/pipeline.ex @@ -531,7 +531,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.Pipeline do end end - @spec maybe_requeue_failed(backend_id :: pos_integer(), messages :: [Message.t()]) :: :ok + @spec maybe_requeue_failed(backend_id :: non_neg_integer(), messages :: [Message.t()]) :: :ok defp maybe_requeue_failed(backend_id, messages) do payloads = Enum.map(messages, & &1.data) @@ -544,7 +544,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.Pipeline do end @spec requeue_or_shed( - backend_id :: pos_integer(), + backend_id :: non_neg_integer(), retriable :: [EncodedRow.t() | LogEventPointer.t()] ) :: :ok defp requeue_or_shed(_backend_id, []), do: :ok @@ -564,7 +564,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.Pipeline do @typep drop_reason :: :retries_exhausted | :circuit_breaker_open @spec requeue_retriable( - backend_id :: pos_integer(), + backend_id :: non_neg_integer(), retriable :: [EncodedRow.t() | LogEventPointer.t()] ) :: :ok defp requeue_retriable(backend_id, retriable) do @@ -590,7 +590,8 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.Pipeline do :ok end - @spec requeue_payload(pos_integer(), EncodedRow.t() | LogEventPointer.t()) :: requeue_result() + @spec requeue_payload(non_neg_integer(), EncodedRow.t() | LogEventPointer.t()) :: + requeue_result() defp requeue_payload(backend_id, %EncodedRow{pointer: pointer} = encoded) do transfer_retry_payload(backend_id, pointer, encoded) end @@ -602,7 +603,11 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.Pipeline do end end - @spec transfer_retry_payload(pos_integer(), LogEventPointer.t(), EncodedRow.t() | LogEvent.t()) :: + @spec transfer_retry_payload( + non_neg_integer(), + LogEventPointer.t(), + EncodedRow.t() | LogEvent.t() + ) :: requeue_result() defp transfer_retry_payload(backend_id, pointer, payload) do pointer = %{pointer | retries: pointer.retries + 1} @@ -624,7 +629,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.Pipeline do defp requeue_transfer_result({:error, :already_exists}), do: :deduplicated defp requeue_transfer_result({:error, :not_initialized}), do: :queue_unavailable - @spec emit_requeue_deduplicated_telemetry(pos_integer(), non_neg_integer()) :: :ok + @spec emit_requeue_deduplicated_telemetry(non_neg_integer(), non_neg_integer()) :: :ok defp emit_requeue_deduplicated_telemetry(_backend_id, 0), do: :ok defp emit_requeue_deduplicated_telemetry(backend_id, deduplicated_count) do @@ -643,7 +648,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.Pipeline do # A lookup miss means GenerationJanitor dropped the backing generation before the # retry could resolve its event. Bounded and rare in practice, but worth # surfacing since it's otherwise invisible. - @spec emit_requeue_lookup_miss_telemetry(pos_integer(), non_neg_integer()) :: :ok + @spec emit_requeue_lookup_miss_telemetry(non_neg_integer(), non_neg_integer()) :: :ok defp emit_requeue_lookup_miss_telemetry(_backend_id, 0), do: :ok defp emit_requeue_lookup_miss_telemetry(backend_id, missing_count) do @@ -659,7 +664,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.Pipeline do ) end - @spec emit_requeue_queue_unavailable_telemetry(pos_integer(), non_neg_integer()) :: :ok + @spec emit_requeue_queue_unavailable_telemetry(non_neg_integer(), non_neg_integer()) :: :ok defp emit_requeue_queue_unavailable_telemetry(_backend_id, 0), do: :ok defp emit_requeue_queue_unavailable_telemetry(backend_id, dropped_count) do @@ -677,7 +682,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.Pipeline do @spec drop_failed( payloads :: [EncodedRow.t() | LogEventPointer.t()], - backend_id :: pos_integer(), + backend_id :: non_neg_integer(), reason :: drop_reason() ) :: :ok defp drop_failed([], _backend_id, _reason), do: :ok diff --git a/lib/logflare/backends/adaptor/clickhouse_adaptor/query_connection_sup.ex b/lib/logflare/backends/adaptor/clickhouse_adaptor/query_connection_sup.ex index f8c0c5fbf4..12c3f2e830 100644 --- a/lib/logflare/backends/adaptor/clickhouse_adaptor/query_connection_sup.ex +++ b/lib/logflare/backends/adaptor/clickhouse_adaptor/query_connection_sup.ex @@ -156,10 +156,10 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.QueryConnectionSup do Note that this call does block while waiting to dispatch to all nodes. """ - @spec recycle_backend(Backend.t() | pos_integer()) :: %{node() => :ok | {:error, term()}} + @spec recycle_backend(Backend.t() | non_neg_integer()) :: %{node() => :ok | {:error, term()}} def recycle_backend(%Backend{id: backend_id}), do: recycle_backend(backend_id) - def recycle_backend(backend_id) when is_pos_integer(backend_id) do + def recycle_backend(backend_id) when is_non_negative_integer(backend_id) do Logger.info("Recycling ClickHouse read pool connections across the cluster", backend_id: backend_id ) @@ -190,8 +190,8 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.QueryConnectionSup do Invoked on each node by `recycle_backend/1`. """ - @spec recycle_backend_local(pos_integer()) :: :ok | {:error, term()} - def recycle_backend_local(backend_id) when is_pos_integer(backend_id) do + @spec recycle_backend_local(non_neg_integer()) :: :ok | {:error, term()} + def recycle_backend_local(backend_id) when is_non_negative_integer(backend_id) do case lookup_managers(backend_id) do [] -> {:error, :no_manager} manager_pids -> manager_pids |> Enum.map(&safe_recycle/1) |> aggregate_results() @@ -211,10 +211,10 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.QueryConnectionSup do Pools restart lazily on demand. """ - @spec refresh_backend(Backend.t() | pos_integer()) :: :ok + @spec refresh_backend(Backend.t() | non_neg_integer()) :: :ok def refresh_backend(%Backend{id: backend_id}), do: refresh_backend(backend_id) - def refresh_backend(backend_id) when is_pos_integer(backend_id) do + def refresh_backend(backend_id) when is_non_negative_integer(backend_id) do Logger.info("Refreshing ClickHouse read pools across the cluster", backend_id: backend_id ) @@ -228,8 +228,8 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.QueryConnectionSup do Invoked on each node by `refresh_backend/1`. """ - @spec refresh_backend_local(pos_integer()) :: :ok - def refresh_backend_local(backend_id) when is_pos_integer(backend_id) do + @spec refresh_backend_local(non_neg_integer()) :: :ok + def refresh_backend_local(backend_id) when is_non_negative_integer(backend_id) do ContextCache.bust_keys([{Backends, backend_id}]) backend_id @@ -250,10 +250,10 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.QueryConnectionSup do Terminates the read `ConnectionManager` and its pool for a backend on every node in the cluster. Used when a backend is deleted. Fire-and-forget. """ - @spec terminate_backend(Backend.t() | pos_integer()) :: :ok + @spec terminate_backend(Backend.t() | non_neg_integer()) :: :ok def terminate_backend(%Backend{id: backend_id}), do: terminate_backend(backend_id) - def terminate_backend(backend_id) when is_pos_integer(backend_id) do + def terminate_backend(backend_id) when is_non_negative_integer(backend_id) do Logger.info("Terminating ClickHouse read connection managers across the cluster", backend_id: backend_id ) @@ -267,8 +267,8 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.QueryConnectionSup do Invoked on each node by `terminate_backend/1`. """ - @spec terminate_backend_local(pos_integer()) :: :ok - def terminate_backend_local(backend_id) when is_pos_integer(backend_id) do + @spec terminate_backend_local(non_neg_integer()) :: :ok + def terminate_backend_local(backend_id) when is_non_negative_integer(backend_id) do backend_id |> lookup_managers() |> Enum.each(&terminate_manager/1) @@ -276,8 +276,8 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.QueryConnectionSup do :ok end - @spec lookup_managers(pos_integer()) :: [pid()] - defp lookup_managers(backend_id) when is_pos_integer(backend_id) do + @spec lookup_managers(non_neg_integer()) :: [pid()] + defp lookup_managers(backend_id) when is_non_negative_integer(backend_id) do manager = ConnectionManager ms = diff --git a/lib/logflare/backends/buffer_producer.ex b/lib/logflare/backends/buffer_producer.ex index d0cf3bd89e..132d72d428 100644 --- a/lib/logflare/backends/buffer_producer.ex +++ b/lib/logflare/backends/buffer_producer.ex @@ -22,7 +22,7 @@ defmodule Logflare.Backends.BufferProducer do demand: non_neg_integer(), source_id: pos_integer() | nil, source_token: atom() | nil, - backend_id: pos_integer(), + backend_id: non_neg_integer(), last_discard_log_dt: DateTime.t() | nil, interval: pos_integer(), in_flight_ref: :atomics.atomics_ref() | nil, @@ -47,7 +47,7 @@ defmodule Logflare.Backends.BufferProducer do @type state :: standard_state() | spool_producer_state() - @type table_key :: {pos_integer() | atom(), pos_integer() | nil, pid() | nil} + @type table_key :: {pos_integer() | atom(), non_neg_integer() | nil, pid() | nil} @in_flight_registry Logflare.Backends.BufferProducer.InFlightRegistry @@ -233,7 +233,7 @@ defmodule Logflare.Backends.BufferProducer do reschedule(state, false, false) end - @spec init_consolidated_state(pos_integer(), pos_integer(), keyword()) :: state() + @spec init_consolidated_state(non_neg_integer(), pos_integer(), keyword()) :: state() defp init_consolidated_state(backend_id, interval, opts) do id_passing = Keyword.get(opts, :id_passing, false) {in_flight_ref, max_in_flight} = init_in_flight(id_passing, Keyword.get(opts, :max_in_flight)) diff --git a/lib/logflare/backends/consolidated_sup.ex b/lib/logflare/backends/consolidated_sup.ex index 6902ae2571..655bf64913 100644 --- a/lib/logflare/backends/consolidated_sup.ex +++ b/lib/logflare/backends/consolidated_sup.ex @@ -57,7 +57,7 @@ defmodule Logflare.Backends.ConsolidatedSup do @doc """ Stops a consolidated pipeline for a backend. """ - @spec stop_pipeline(Backend.t() | pos_integer()) :: :ok | {:error, :not_found} + @spec stop_pipeline(Backend.t() | non_neg_integer()) :: :ok | {:error, :not_found} def stop_pipeline(%Backend{} = backend) do case find_pipeline_pid(backend) do nil -> @@ -81,7 +81,7 @@ defmodule Logflare.Backends.ConsolidatedSup do @doc """ Checks if a consolidated pipeline is running for a backend. """ - @spec pipeline_running?(Backend.t() | pos_integer()) :: boolean() + @spec pipeline_running?(Backend.t() | non_neg_integer()) :: boolean() def pipeline_running?(%Backend{} = backend) do find_pipeline_pid(backend) != nil end diff --git a/lib/logflare/backends/consolidated_sup_worker.ex b/lib/logflare/backends/consolidated_sup_worker.ex index 25cff8333a..b1d7d1496e 100644 --- a/lib/logflare/backends/consolidated_sup_worker.ex +++ b/lib/logflare/backends/consolidated_sup_worker.ex @@ -14,6 +14,7 @@ defmodule Logflare.Backends.ConsolidatedSupWorker do alias Logflare.Backends alias Logflare.Backends.Adaptor alias Logflare.Backends.ConsolidatedSup + alias Logflare.SingleTenant @default_interval 30_000 @@ -52,9 +53,19 @@ defmodule Logflare.Backends.ConsolidatedSupWorker do end defp list_expected_backend_ids do - Backends.list_backends(has_sources_or_rules: true) - |> Enum.filter(&Adaptor.consolidated_ingest?/1) - |> Enum.map(& &1.id) + default_backend_ids = + case SingleTenant.get_default_clickhouse_backend() do + nil -> [] + backend -> [backend.id] + end + + backend_ids = + Backends.list_backends(has_sources_or_rules: true) + |> Enum.filter(&Adaptor.consolidated_ingest?/1) + |> Enum.map(& &1.id) + + default_backend_ids + |> Enum.concat(backend_ids) |> MapSet.new() end diff --git a/lib/logflare/backends/ingest_event_queue.ex b/lib/logflare/backends/ingest_event_queue.ex index abb41b2026..8e043c153b 100644 --- a/lib/logflare/backends/ingest_event_queue.ex +++ b/lib/logflare/backends/ingest_event_queue.ex @@ -46,8 +46,8 @@ defmodule Logflare.Backends.IngestEventQueue do @type table_key :: {pos_integer(), pos_integer() | nil, pid() | nil} @type table_obj :: {table_key(), :ets.tid()} @type queues_key :: {pos_integer(), pos_integer() | nil} - @type consolidated_queues_key :: {:consolidated, pos_integer()} - @type consolidated_table_key :: {:consolidated, pos_integer(), pid() | nil} + @type consolidated_queues_key :: {:consolidated, non_neg_integer()} + @type consolidated_table_key :: {:consolidated, non_neg_integer(), pid() | nil} @type spool_producer_queues_key :: {:spool_producer, nil} @type spool_producer_table_key :: {:spool_producer, nil, pid() | nil} @type pointer_batch_key :: diff --git a/lib/logflare/backends/ingest_event_queue/queue_janitor.ex b/lib/logflare/backends/ingest_event_queue/queue_janitor.ex index e0cdac926e..e4fc5c4fee 100644 --- a/lib/logflare/backends/ingest_event_queue/queue_janitor.ex +++ b/lib/logflare/backends/ingest_event_queue/queue_janitor.ex @@ -88,7 +88,7 @@ defmodule Logflare.Backends.IngestEventQueue.QueueJanitor do @spec drop_queue( map(), {pos_integer(), pos_integer() | nil, pid() | nil} - | {:consolidated, pos_integer(), pid() | nil}, + | {:consolidated, non_neg_integer(), pid() | nil}, IngestEventQueue.queues_key() | IngestEventQueue.consolidated_queues_key(), pid() | nil, boolean(), diff --git a/lib/logflare/backends/source_sup.ex b/lib/logflare/backends/source_sup.ex index 43f8117214..efc6caa981 100644 --- a/lib/logflare/backends/source_sup.ex +++ b/lib/logflare/backends/source_sup.ex @@ -103,11 +103,9 @@ defmodule Logflare.Backends.SourceSup do defp init_children(source) do ingest_backends = Backends.Cache.list_backends(source_id: source.id) - |> Enum.reject(& &1.consolidated_ingest?) rules_backends = Backends.Cache.list_backends(rules_source_id: source.id) - |> Enum.reject(& &1.consolidated_ingest?) |> Enum.map(&%{&1 | register_for_ingest: false}) user = Users.Cache.get(source.user_id) @@ -119,6 +117,8 @@ defmodule Logflare.Backends.SourceSup do specs = [default_backend | ingest_backends] |> Enum.concat(rules_backends) + |> Enum.reject(&is_nil/1) + |> Enum.reject(& &1.consolidated_ingest?) |> Enum.map(&Backend.child_spec(source, &1)) |> Enum.uniq() diff --git a/lib/logflare/backends/system_backend.ex b/lib/logflare/backends/system_backend.ex new file mode 100644 index 0000000000..ad5807b2ce --- /dev/null +++ b/lib/logflare/backends/system_backend.ex @@ -0,0 +1,45 @@ +defmodule Logflare.Backends.SystemBackend do + @moduledoc """ + Lifecycle hooks for the system backend, distinct from user-configured backends. + + Hooks are optional; adaptors without a hook require no additional setup. + """ + + alias Logflare.Backends.Adaptor + alias Logflare.Backends.Backend + alias Logflare.SingleTenant + alias Logflare.Sources.Source + + @doc "Runs before single-tenant users and sources are seeded." + @callback on_system_start() :: :ok + + @doc "Runs after a source supervisor starts." + @callback on_source_start(Source.t()) :: :ok + + @doc "Runs after Supabase sources and endpoints are seeded and source startup is requested." + @callback on_supabase_start() :: :ok + + @optional_callbacks on_system_start: 0, on_source_start: 1, on_supabase_start: 0 + + @spec on_system_start() :: :ok + def on_system_start, do: invoke(:on_system_start, []) + + @spec on_source_start(Source.t()) :: :ok + def on_source_start(%Source{} = source), do: invoke(:on_source_start, [source]) + + @spec on_supabase_start() :: :ok + def on_supabase_start, do: invoke(:on_supabase_start, []) + + @spec invoke(atom(), list()) :: :ok + defp invoke(callback, args) do + type = if SingleTenant.single_tenant?(), do: SingleTenant.backend_type(), else: :bigquery + adaptor = Adaptor.get_adaptor(%Backend{type: type}) + Code.ensure_loaded!(adaptor) + + if function_exported?(adaptor, callback, length(args)) do + apply(adaptor, callback, args) + else + :ok + end + end +end diff --git a/lib/logflare/networking.ex b/lib/logflare/networking.ex index 21e60766b8..9c8fab3f60 100644 --- a/lib/logflare/networking.ex +++ b/lib/logflare/networking.ex @@ -4,7 +4,7 @@ defmodule Logflare.Networking do alias Logflare.Backends alias Logflare.Backends.Adaptor.BigQueryAdaptor.GoogleApiClient alias Logflare.Backends.Adaptor.DatadogAdaptor - alias Logflare.SingleTenant + alias Logflare.Backends # Finch bounds only connect and receive. Without `send_timeout` a peer that accepts # the connection and then stops reading blocks the insert inside `:gen_tcp.send` @@ -20,17 +20,15 @@ defmodule Logflare.Networking do @spool_pool_size 150 def pools do - if SingleTenant.postgres_backend?() do - finch_pools(true) + if Backends.bigquery_default_backend?() do + bigquery_finch_pools() ++ grpc_pools() else - finch_pools(false) ++ grpc_pools() + base_finch_pools_with_default() end end - defp finch_pools(true = _postgres_backend?) do + defp base_finch_pools_with_default do [ - {Finch, - name: Logflare.FinchDefaultHttp1, pools: %{default: [protocols: [:http1], size: 50]}}, {Finch, name: Logflare.FinchDefault, pools: @@ -43,15 +41,13 @@ defmodule Logflare.Networking do ] end - defp finch_pools(false = _postgres_backend?) do + defp bigquery_finch_pools do base = System.schedulers_online() http1_count = max(div(base, 4), 1) [ # Finch connection pools, using http2 {Finch, name: Logflare.FinchGoth, pools: %{default: [protocols: [:http2], count: 1]}}, - {Finch, - name: Logflare.FinchDefaultHttp1, pools: %{default: [protocols: [:http1], size: 50]}}, {Finch, name: Logflare.FinchIngest, pools: %{ @@ -150,6 +146,8 @@ defmodule Logflare.Networking do base = System.schedulers_online() [ + {Finch, + name: Logflare.FinchDefaultHttp1, pools: %{default: [protocols: [:http1], size: 50]}}, {Finch, name: Logflare.FinchClickHouseIngest, pools: %{ diff --git a/lib/logflare/pubsub_rates/cache.ex b/lib/logflare/pubsub_rates/cache.ex index 0bc767f83e..3b7c01a943 100644 --- a/lib/logflare/pubsub_rates/cache.ex +++ b/lib/logflare/pubsub_rates/cache.ex @@ -56,6 +56,7 @@ defmodule Logflare.PubSubRates.Cache do Merges a node map into the local cache. """ @typep node_buffers :: %{atom() => non_neg_integer()} + @spec cache_buffers(non_neg_integer() | :consolidated, non_neg_integer() | nil, node_buffers()) :: {:ok, true} def cache_buffers(:consolidated, backend_id, buffers) when is_integer(backend_id), @@ -97,8 +98,12 @@ defmodule Logflare.PubSubRates.Cache do Returns the sum of all buffers across the cluster for a given source and backend combination. """ @spec get_cluster_buffers(non_neg_integer()) :: non_neg_integer() - @spec get_cluster_buffers(non_neg_integer(), non_neg_integer() | nil) :: non_neg_integer() - def get_cluster_buffers(source_id, backend_id \\ nil) when is_integer(source_id) do + @spec get_cluster_buffers(non_neg_integer() | :consolidated, non_neg_integer() | nil) :: + non_neg_integer() + + def get_cluster_buffers(source_id, backend_id \\ nil) + when is_integer(source_id) or + (source_id == :consolidated and is_integer(backend_id)) do case get_buffers(source_id, backend_id) do {:ok, node_buffers} when node_buffers != nil -> merge_buffers(node_buffers) _ -> 0 diff --git a/lib/logflare/single_tenant.ex b/lib/logflare/single_tenant.ex index e60b969343..8682f56db0 100644 --- a/lib/logflare/single_tenant.ex +++ b/lib/logflare/single_tenant.ex @@ -19,6 +19,8 @@ defmodule Logflare.SingleTenant do alias Logflare.SourceSchemas require Logger + @single_tenant_clickhouse_backend_id 0 + @user_attrs %{ name: "default", email: "default@logflare.app", @@ -55,6 +57,8 @@ defmodule Logflare.SingleTenant do "pgbouncer.logs.prod" ] + @type backend_type :: :bigquery | :clickhouse | :postgres + defp endpoint_params do [ %{ @@ -109,6 +113,40 @@ defmodule Logflare.SingleTenant do |> Backends.get_default_backend() end + @doc """ + Builds the synthetic backend when ClickHouse is selected in single-tenant mode. + + Uses the supplied user's ID, looking up the default user only when omitted. + Returns nil when ClickHouse is not selected or the user is nil. + + Backend ID `0` is reserved for this synthetic backend and identifies its + consolidated pipeline and ingestion queues. The backend is not persisted. + """ + @spec get_default_clickhouse_backend(User.t() | nil) :: Backend.t() | nil + @spec get_default_clickhouse_backend() :: Backend.t() | nil + def get_default_clickhouse_backend(user \\ get_default_user()) + + def get_default_clickhouse_backend(nil), do: nil + + def get_default_clickhouse_backend(%User{id: user_id}) do + if clickhouse_backend?() do + %Backend{ + id: @single_tenant_clickhouse_backend_id, + type: :clickhouse, + config: Map.new(clickhouse_backend_adapter_opts()), + user_id: user_id, + name: "Default ClickHouse backend", + consolidated_ingest?: true + } + end + end + + @doc "Returns a source's synthetic ClickHouse backend, or nil for legacy per-source defaults." + @spec lookup_system_default_backend(Source.t()) :: Backend.t() | nil + def lookup_system_default_backend(%Source{user_id: user_id}) do + get_default_clickhouse_backend(%User{id: user_id}) + end + @doc """ Creates an enterprise user """ @@ -125,6 +163,17 @@ defmodule Logflare.SingleTenant do end end + @doc "Ensures the selected single-tenant ClickHouse pipeline is running" + @spec ensure_default_backend() :: :ok + def ensure_default_backend do + if clickhouse_backend?() do + get_default_backend() + |> Backends.maybe_restart_consolidated_pipeline() + end + + :ok + end + @doc """ Create access tokens """ @@ -261,10 +310,10 @@ defmodule Logflare.SingleTenant do end defp start_supabase_source(source) do - if postgres_backend?() do - Backends.start_source_sup(source) - else + if bigquery_backend?() do Supervisor.ensure_started(source) + else + Backends.start_source_sup(source) end end @@ -294,18 +343,47 @@ defmodule Logflare.SingleTenant do @spec single_tenant? :: boolean() def single_tenant?, do: !!Application.get_env(:logflare, :single_tenant) + @doc "Builds ClickHouse adapter configuration from a connection URL" + @spec clickhouse_backend_adapter_opts_from_url!(String.t()) :: keyword() + def clickhouse_backend_adapter_opts_from_url!(url) when is_binary(url) do + uri = parse_clickhouse_backend_url!(url) + {username, password} = clickhouse_credentials(uri) + + sanitized_url = + uri + |> Map.merge(%{userinfo: nil, path: nil, query: nil, fragment: nil}) + |> URI.to_string() + + [ + url: sanitized_url, + username: username, + password: password, + database: clickhouse_database(uri), + port: clickhouse_port(url, uri.scheme) + ] + |> Enum.reject(fn {_key, value} -> is_nil(value) end) + end + @doc "Returns true if supabase mode flag is set via config and if is single tenant" @spec supabase_mode? :: boolean() def supabase_mode?, do: !!Application.get_env(:logflare, :supabase_mode) and single_tenant?() - @doc "Returns postgres backend adapter configurations if single tenant and env is set" - @spec postgres_backend_adapter_opts :: Keyword.t() | nil + @doc "Returns PostgreSQL backend adapter configuration in single-tenant mode" + @spec postgres_backend_adapter_opts :: keyword() | nil def postgres_backend_adapter_opts do if single_tenant?() do Application.get_env(:logflare, :postgres_backend_adapter) end end + @doc "Returns ClickHouse backend adapter configuration when selected in single-tenant mode" + @spec clickhouse_backend_adapter_opts :: keyword() | nil + def clickhouse_backend_adapter_opts do + if clickhouse_backend?() do + Application.get_env(:logflare, :clickhouse_backend_adapter) + end + end + @doc """ Adds ingestion samples for supabase sources, so that schema is built and stored correctly. """ @@ -359,7 +437,7 @@ defmodule Logflare.SingleTenant do source_schemas_updated = cond do - postgres_backend?() -> :ok + not bigquery_backend?() -> :ok # for mocking, use full qualified function name __MODULE__.supabase_mode_source_schemas_updated?() -> :ok true -> nil @@ -375,27 +453,34 @@ defmodule Logflare.SingleTenant do end @doc """ - Returns true if postgres backend is enabled for single tenant + Returns the selected single-tenant backend type. """ + @spec backend_type :: backend_type() + def backend_type do + Application.fetch_env!(:logflare, :single_tenant_backend) + end + + @doc "Returns true if BigQuery is selected in single-tenant mode" + @spec bigquery_backend? :: boolean() + def bigquery_backend?, do: selected_backend?(:bigquery) + + @doc "Returns true if PostgreSQL is selected in single-tenant mode" @spec postgres_backend? :: boolean() - def postgres_backend? do - opts = postgres_backend_adapter_opts() || [] + def postgres_backend?, do: selected_backend?(:postgres) - url = Keyword.get(opts, :url) - single_tenant?() && url != nil - end + @doc "Returns true if ClickHouse is selected in single-tenant mode" + @spec clickhouse_backend? :: boolean() + def clickhouse_backend?, do: selected_backend?(:clickhouse) - @doc """ - Returns the type of the default backend - """ - @spec backend_type :: :postgres | :bigquery - def backend_type do - if postgres_backend?() do - :postgres - else - :bigquery - end - end + @doc "Returns whether this is the default backend in a single-tenant ClickHouse mode." + @spec default_clickhouse_backend?(Backend.t()) :: boolean() + def default_clickhouse_backend?(%Backend{ + id: @single_tenant_clickhouse_backend_id, + type: :clickhouse + }), + do: clickhouse_backend?() + + def default_clickhouse_backend?(%Backend{}), do: false def supabase_mode_source_schemas_updated? do user = get_default_user() @@ -423,4 +508,40 @@ defmodule Logflare.SingleTenant do |> File.read!() |> Jason.decode!() end + + defp parse_clickhouse_backend_url!(url) do + case URI.parse(url) do + %URI{host: host} = uri when host not in [nil, ""] -> uri + %URI{} -> raise "CLICKHOUSE_BACKEND_URL must include a hostname" + end + end + + defp clickhouse_database(%URI{path: path}) when path in [nil, "", "/"], do: "default" + + defp clickhouse_database(%URI{path: path}) do + path + |> String.trim_leading("/") + |> URI.decode() + end + + defp clickhouse_credentials(%URI{userinfo: nil}), do: {nil, nil} + + defp clickhouse_credentials(%URI{userinfo: userinfo}) do + case String.split(userinfo, ":", parts: 2) do + [username, password] -> {URI.decode(username), URI.decode(password)} + [username] -> {URI.decode(username), nil} + end + end + + defp clickhouse_port(url, scheme) do + case :uri_string.parse(url) do + %{port: port} -> port + _uri when scheme == "https" -> 8443 + _uri -> 8123 + end + end + + defp selected_backend?(backend_type) do + single_tenant?() and backend_type() == backend_type + end end diff --git a/lib/logflare/sources.ex b/lib/logflare/sources.ex index e2aacdd871..eb30b3e6b4 100644 --- a/lib/logflare/sources.ex +++ b/lib/logflare/sources.ex @@ -15,7 +15,6 @@ defmodule Logflare.Sources do alias Logflare.Logs.RejectedLogEvents alias Logflare.PubSubRates alias Logflare.Repo - alias Logflare.SingleTenant alias Logflare.SourceSchemas alias Logflare.Sources alias Logflare.Sources.Source @@ -132,7 +131,7 @@ defmodule Logflare.Sources do |> Ecto.build_assoc(:sources) |> Source.update_by_user_changeset(source_params) |> Repo.insert() do - if !SingleTenant.postgres_backend?() do + if Backends.bigquery_default_backend?() do create_big_query_schema_and_start_source(source) end @@ -249,7 +248,8 @@ defmodule Logflare.Sources do source = put_retention_days(source) updated = put_retention_days(updated) - if source.retention_days != updated.retention_days and not SingleTenant.postgres_backend?() do + if source.retention_days != updated.retention_days and + Backends.bigquery_default_backend?() do user = Users.Cache.get(updated.user_id) BigQuery.patch_table_ttl( @@ -261,7 +261,7 @@ defmodule Logflare.Sources do end if source.bigquery_clustering_fields != updated.bigquery_clustering_fields and - not SingleTenant.postgres_backend?() do + Backends.bigquery_default_backend?() do user = Users.Cache.get(updated.user_id) fields = String.split(updated.bigquery_clustering_fields || "", ",") ++ ["timestamp"] @@ -666,26 +666,24 @@ defmodule Logflare.Sources do end @spec calculate_total_pending(Source.t()) :: non_neg_integer() - defp calculate_total_pending(%Source{id: source_id} = source) do - # Get all backends for this source, including default backend - user = Users.Cache.get(source.user_id) - default_backend = Backends.get_default_backend(user) - + defp calculate_total_pending(%Source{} = source) do ingest_backends = Backends.Cache.list_backends(source_id: source.id) rules_backends = Backends.Cache.list_backends(rules_source_id: source.id) |> Enum.map(&%{&1 | register_for_ingest: false}) - all_backends = [default_backend | ingest_backends] ++ rules_backends + backend_queue_keys = + (ingest_backends ++ rules_backends) + |> Enum.map(&Backends.ingest_queue_key(source, &1)) + + queue_keys = [Backends.system_default_ingest_queue_key(source) | backend_queue_keys] # Sum pending items across all backends - all_backends - |> Enum.reduce(0, fn backend, acc -> - case Backends.IngestEventQueue.total_pending({source_id, backend.id}) do - count -> acc + count - end - end) + queue_keys + |> Enum.reject(&is_nil/1) + |> Enum.uniq() + |> Enum.reduce(0, &(Backends.IngestEventQueue.total_pending(&1) + &2)) end @spec has_recent_logs_within?(Source.t(), non_neg_integer()) :: boolean() diff --git a/lib/logflare/sources/source/rate_counter_server.ex b/lib/logflare/sources/source/rate_counter_server.ex index 9b0e3ad1bd..80d3f47336 100644 --- a/lib/logflare/sources/source/rate_counter_server.ex +++ b/lib/logflare/sources/source/rate_counter_server.ex @@ -8,7 +8,6 @@ defmodule Logflare.Sources.Source.RateCounterServer do alias Logflare.Backends alias Logflare.Google.BigQuery.GenUtils alias Logflare.PubSubRates - alias Logflare.SingleTenant alias Logflare.Source.Data alias Logflare.Sources alias Logflare.Sources.Counters @@ -64,7 +63,7 @@ defmodule Logflare.Sources.Source.RateCounterServer do put_current_rate() bigquery_project_id = - if !SingleTenant.postgres_backend?() do + if Backends.bigquery_default_backend?() do GenUtils.get_project_id(source_token) end diff --git a/lib/logflare/sources/source/supervisor.ex b/lib/logflare/sources/source/supervisor.ex index f37eeb4195..a6d17f4b7c 100644 --- a/lib/logflare/sources/source/supervisor.ex +++ b/lib/logflare/sources/source/supervisor.ex @@ -5,18 +5,15 @@ defmodule Logflare.Sources.Source.Supervisor do use GenServer alias Logflare.Backends + alias Logflare.Backends.SystemBackend alias Logflare.ContextCache alias Logflare.Google.BigQuery alias Logflare.Repo alias Logflare.SourceSchemas alias Logflare.Sources alias Logflare.Sources.Counters - alias Logflare.Google.BigQuery - alias Logflare.ContextCache - alias Logflare.SourceSchemas - alias Logflare.Backends - alias Logflare.Utils.Tasks alias Logflare.Sources.Source + require Logger # TODO: Move all manager fns into a manager server so errors in manager fns don't kill the whole supervision tree @@ -99,7 +96,6 @@ defmodule Logflare.Sources.Source.Supervisor do ## Public Functions def start_source(source_token) when is_atom(source_token) do - # Calling this server doing boot times out due to dealing with bigquery in init_table() GenServer.abcast(__MODULE__, {:create, source_token}) {:ok, source_token} @@ -108,7 +104,7 @@ defmodule Logflare.Sources.Source.Supervisor do def delete_source(source_token) do GenServer.abcast(__MODULE__, {:stop, source_token}) # TODO: move to adaptor callback - unless do_pg_ops?() do + if Backends.bigquery_default_backend?() do BigQuery.delete_table(source_token) end @@ -121,7 +117,7 @@ defmodule Logflare.Sources.Source.Supervisor do end def reset_source(source_token) do - unless do_pg_ops?() do + if Backends.bigquery_default_backend?() do GenServer.abcast(__MODULE__, {:restart, source_token}) end @@ -140,14 +136,9 @@ defmodule Logflare.Sources.Source.Supervisor do |> Enum.each(fn s -> reset_source(s.token) end) end - defp do_pg_ops? do - !!Application.get_env(:logflare, :single_tenant) && - !!Application.get_env(:logflare, :postgres_backend_adapter) - end - defp create_source(%Source{} = source) do with {:ok, _pid} = res <- do_start_source_sup(source), - :ok <- init_table(source.token) do + :ok <- SystemBackend.on_source_start(source) do res else {:error, :already_started} = err -> @@ -157,7 +148,7 @@ defmodule Logflare.Sources.Source.Supervisor do {:error, reason} {:error, :start_timeout} = err -> - init_table(source.token) + SystemBackend.on_source_start(source) err {:error, _reason} = err -> @@ -183,29 +174,6 @@ defmodule Logflare.Sources.Source.Supervisor do :ok end - def init_table(source_token) do - %{ - user_id: user_id, - bigquery_table_ttl: bigquery_table_ttl, - bigquery_project_id: bigquery_project_id, - bigquery_dataset_location: bigquery_dataset_location, - bigquery_dataset_id: bigquery_dataset_id - } = BigQuery.GenUtils.get_bq_user_info(source_token) - - Tasks.start_child(fn -> - BigQuery.init_table!( - user_id, - source_token, - bigquery_project_id, - bigquery_table_ttl, - bigquery_dataset_location, - bigquery_dataset_id - ) - end) - - :ok - end - defp do_start_source_sup(source) do with :ok <- Backends.start_source_sup(source) do do_lookup(source) diff --git a/test/config/runtime_test.exs b/test/config/runtime_test.exs new file mode 100644 index 0000000000..bd914e6c49 --- /dev/null +++ b/test/config/runtime_test.exs @@ -0,0 +1,120 @@ +defmodule Logflare.RuntimeConfigTest do + use ExUnit.Case, async: false + + alias Logflare.SingleTenant + + @backend_env_vars [ + "GOOGLE_SERVICE_ACCOUNT", + "POSTGRES_BACKEND_URL", + "CLICKHOUSE_BACKEND_URL", + "LOGFLARE_CLICKHOUSE_TABLE_SUFFIX" + ] + @clickhouse_backend_url "http://localhost:8123/logflare" + @postgres_backend_url "postgresql://postgres:postgres@localhost/logflare" + + setup do + previous_values = + ["LOGFLARE_SINGLE_TENANT" | @backend_env_vars] + |> Map.new(&{&1, System.get_env(&1)}) + + previous_backend = Application.fetch_env(:logflare, :single_tenant_backend) + + System.put_env("LOGFLARE_SINGLE_TENANT", "true") + Enum.each(@backend_env_vars, &System.delete_env/1) + + on_exit(fn -> + case previous_backend do + {:ok, backend} -> Application.put_env(:logflare, :single_tenant_backend, backend) + :error -> Application.delete_env(:logflare, :single_tenant_backend) + end + + Enum.each(previous_values, fn + {name, nil} -> System.delete_env(name) + {name, value} -> System.put_env(name, value) + end) + end) + end + + test "defaults the single-tenant backend to BigQuery when no backend URL is configured" do + assert runtime_backend_type() == :bigquery + end + + test "configures the ClickHouse table suffix" do + assert get_in(runtime_config(), [:logflare, :clickhouse_table_suffix]) == "default" + + System.put_env("LOGFLARE_CLICKHOUSE_TABLE_SUFFIX", "custom") + + assert get_in(runtime_config(), [:logflare, :clickhouse_table_suffix]) == "custom" + end + + test "selects the configured backend" do + for {env, value, expected_backend} <- [ + {"GOOGLE_SERVICE_ACCOUNT", "service-account", :bigquery}, + {"POSTGRES_BACKEND_URL", @postgres_backend_url, :postgres}, + {"CLICKHOUSE_BACKEND_URL", @clickhouse_backend_url, :clickhouse} + ] do + System.put_env(env, value) + + assert runtime_backend_type() == expected_backend + + System.delete_env(env) + end + end + + test "ignores blank backend URLs" do + System.put_env("POSTGRES_BACKEND_URL", " ") + System.put_env("CLICKHOUSE_BACKEND_URL", "\t") + + assert runtime_backend_type() == :bigquery + end + + test "rejects multiple backend URL configurations" do + System.put_env("POSTGRES_BACKEND_URL", @postgres_backend_url) + System.put_env("CLICKHOUSE_BACKEND_URL", @clickhouse_backend_url) + + assert_raise RuntimeError, + "Only one of GOOGLE_SERVICE_ACCOUNT, POSTGRES_BACKEND_URL, or CLICKHOUSE_BACKEND_URL may be configured", + &runtime_config/0 + end + + test "rejects BigQuery with another backend configuration" do + System.put_env("GOOGLE_SERVICE_ACCOUNT", "service-account") + + for {backend_env, value} <- [ + {"POSTGRES_BACKEND_URL", @postgres_backend_url}, + {"CLICKHOUSE_BACKEND_URL", @clickhouse_backend_url} + ] do + System.put_env(backend_env, value) + + assert_raise RuntimeError, + "Only one of GOOGLE_SERVICE_ACCOUNT, POSTGRES_BACKEND_URL, or CLICKHOUSE_BACKEND_URL may be configured", + &runtime_config/0 + + System.delete_env(backend_env) + end + end + + test "defaults to BigQuery outside single-tenant mode" do + System.put_env("LOGFLARE_SINGLE_TENANT", "false") + + for {configured_env, value, unconfigured_env} <- [ + {"POSTGRES_BACKEND_URL", @postgres_backend_url, "CLICKHOUSE_BACKEND_URL"}, + {"CLICKHOUSE_BACKEND_URL", @clickhouse_backend_url, "POSTGRES_BACKEND_URL"} + ] do + System.put_env(configured_env, value) + System.delete_env(unconfigured_env) + + assert runtime_backend_type() == :bigquery + end + end + + defp runtime_backend_type do + backend_type = get_in(runtime_config(), [:logflare, :single_tenant_backend]) + Application.put_env(:logflare, :single_tenant_backend, backend_type) + SingleTenant.backend_type() + end + + defp runtime_config do + Config.Reader.read!("config/runtime.exs", env: :test) + end +end diff --git a/test/logflare/backends/adaptor/clickhouse_adaptor/connection_manager_test.exs b/test/logflare/backends/adaptor/clickhouse_adaptor/connection_manager_test.exs index 16e92e13bf..c5752084f7 100644 --- a/test/logflare/backends/adaptor/clickhouse_adaptor/connection_manager_test.exs +++ b/test/logflare/backends/adaptor/clickhouse_adaptor/connection_manager_test.exs @@ -285,10 +285,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.ConnectionManagerTest do describe "query pool credentials" do test "the pool authenticates with the dedicated query user when configured" do {_source, backend} = - setup_clickhouse_test( - config: %{query_user: "ch_reader", query_password: "reader_pa55"}, - cleanup?: false - ) + setup_clickhouse_test(config: %{query_user: "ch_reader", query_password: "reader_pa55"}) assert {"ch_reader", "reader_pa55"} = start_pool_credentials(backend) end @@ -301,7 +298,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.ConnectionManagerTest do test "the pool falls back to the default credentials when only query_user is configured" do {_source, backend} = - setup_clickhouse_test(config: %{query_user: "ch_reader"}, cleanup?: false) + setup_clickhouse_test(config: %{query_user: "ch_reader"}) assert {"logflare", "logflare"} = start_pool_credentials(backend) end diff --git a/test/logflare/backends/adaptor/clickhouse_adaptor/ingester_test.exs b/test/logflare/backends/adaptor/clickhouse_adaptor/ingester_test.exs index da1ff2fff4..f76869e05a 100644 --- a/test/logflare/backends/adaptor/clickhouse_adaptor/ingester_test.exs +++ b/test/logflare/backends/adaptor/clickhouse_adaptor/ingester_test.exs @@ -389,12 +389,8 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.IngesterTest do insert(:plan, name: "Free") {source, backend} = setup_clickhouse_test() - {:ok, _supervisor_pid} = ClickHouseAdaptor.start_link(backend) - table_name = ClickHouseAdaptor.clickhouse_ingest_table_name(backend, :log) - Process.sleep(200) - [source: source, backend: backend, table_name: table_name] end diff --git a/test/logflare/backends/adaptor/clickhouse_adaptor/pipeline_test.exs b/test/logflare/backends/adaptor/clickhouse_adaptor/pipeline_test.exs index 3546c42ee6..75fc104fe3 100644 --- a/test/logflare/backends/adaptor/clickhouse_adaptor/pipeline_test.exs +++ b/test/logflare/backends/adaptor/clickhouse_adaptor/pipeline_test.exs @@ -27,18 +27,11 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.PipelineTest do insert(:plan, name: "Free") {source, backend} = setup_clickhouse_test() + drop_clickhouse_tables_on_exit(backend) - {:ok, supervisor_pid} = ClickHouseAdaptor.start_link(backend) + start_supervised!({ClickHouseAdaptor, backend}) - on_exit(fn -> - if Process.alive?(supervisor_pid) do - Process.exit(supervisor_pid, :shutdown) - end - end) - - TestUtils.retry_assert(fn -> - assert :ok = ClickHouseAdaptor.provision_ingest_tables(backend) - end) + assert :ok = ClickHouseAdaptor.provision_ingest_tables(backend) context = Pipeline.build_processor_context(backend.id) diff --git a/test/logflare/backends/adaptor/clickhouse_adaptor/provisioner_test.exs b/test/logflare/backends/adaptor/clickhouse_adaptor/provisioner_test.exs index 6fcc3bfcba..fccb2bcc0c 100644 --- a/test/logflare/backends/adaptor/clickhouse_adaptor/provisioner_test.exs +++ b/test/logflare/backends/adaptor/clickhouse_adaptor/provisioner_test.exs @@ -11,14 +11,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.ProvisionerTest do insert(:plan, name: "Free") {source, backend} = setup_clickhouse_test() - - {:ok, supervisor_pid} = ClickHouseAdaptor.start_link(backend) - - on_exit(fn -> - if Process.alive?(supervisor_pid) do - Process.exit(supervisor_pid, :shutdown) - end - end) + drop_clickhouse_tables_on_exit(backend) [source: source, backend: backend] end @@ -94,13 +87,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.ProvisionerTest do {_source, read_backend} = setup_clickhouse_test(config: %{query_user: "ch_reader", query_password: "reader_pa55"}) - {:ok, adaptor_pid} = ClickHouseAdaptor.start_link(read_backend) - - on_exit(fn -> - if Process.alive?(adaptor_pid) do - Process.exit(adaptor_pid, :shutdown) - end - end) + drop_clickhouse_tables_on_exit(read_backend) {:ok, pid} = Provisioner.start_link(read_backend) ref = Process.monitor(pid) diff --git a/test/logflare/backends/adaptor/clickhouse_adaptor/single_tenant_ingestion_test.exs b/test/logflare/backends/adaptor/clickhouse_adaptor/single_tenant_ingestion_test.exs new file mode 100644 index 0000000000..f6828c9fae --- /dev/null +++ b/test/logflare/backends/adaptor/clickhouse_adaptor/single_tenant_ingestion_test.exs @@ -0,0 +1,47 @@ +defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.SingleTenantIngestionTest do + use Logflare.DataCase, async: false + + alias Logflare.Backends.Adaptor.ClickHouseAdaptor + alias Logflare.Backends.Adaptor.ClickHouseAdaptor.Provisioner + alias Logflare.SingleTenant + alias Logflare.SystemMetrics.AllLogsLogged + + TestUtils.setup_single_tenant(backend_type: :clickhouse, seed_user: true) + + setup do + insert(:plan, name: "Free") + + user = SingleTenant.get_default_user() + source = insert(:source, user: user) + backend = Logflare.Backends.get_default_backend(user) + + drop_clickhouse_tables_on_exit(backend) + start_supervised!(AllLogsLogged) + start_supervised!({ClickHouseAdaptor, backend}) + + [source: source, backend: backend] + end + + test "the pipeline consumes normally ingested events", %{ + source: source, + backend: backend + } do + {:ok, provisioner_pid} = Provisioner.start_link(backend) + provisioner_ref = Process.monitor(provisioner_pid) + assert_receive {:DOWN, ^provisioner_ref, :process, ^provisioner_pid, :normal}, 5_000 + + message = "synthetic pipeline #{System.unique_integer([:positive])}" + assert {:ok, 1} = Logflare.Backends.ingest_logs([%{"message" => message}], source) + + table_name = ClickHouseAdaptor.clickhouse_ingest_table_name(backend, :log) + + TestUtils.retry_assert([duration: 10_000], fn -> + assert {:ok, {[%{"count" => 1}], _bytes}} = + ClickHouseAdaptor.execute_ch_query( + backend, + "SELECT count(*) AS count FROM #{table_name} WHERE event_message = {message:String}", + %{message: message} + ) + end) + end +end diff --git a/test/logflare/backends/adaptor/clickhouse_adaptor_test.exs b/test/logflare/backends/adaptor/clickhouse_adaptor_test.exs index ed43b09616..7421ca3adc 100644 --- a/test/logflare/backends/adaptor/clickhouse_adaptor_test.exs +++ b/test/logflare/backends/adaptor/clickhouse_adaptor_test.exs @@ -24,16 +24,26 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do doctest ClickHouseAdaptor describe "table name generation" do + TestUtils.setup_single_tenant(backend_type: :clickhouse) + setup do - insert(:plan, name: "Free") + previous_table_suffix = Application.fetch_env(:logflare, :clickhouse_table_suffix) + Application.put_env(:logflare, :clickhouse_table_suffix, "default") - {source, backend} = setup_clickhouse_test() + on_exit(fn -> + case previous_table_suffix do + {:ok, suffix} -> Application.put_env(:logflare, :clickhouse_table_suffix, suffix) + :error -> Application.delete_env(:logflare, :clickhouse_table_suffix) + end + end) + + backend = %Backend{type: :clickhouse, token: Ecto.UUID.generate()} stringified_backend_token = backend.token |> String.replace("-", "_") - [source: source, backend: backend, stringified_backend_token: stringified_backend_token] + [backend: backend, stringified_backend_token: stringified_backend_token] end test "raises when table name is equal to or exceeds 200 chars", @@ -58,6 +68,54 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do assert ClickHouseAdaptor.clickhouse_ingest_table_name(backend, :trace) == "otel_traces_#{stringified_backend_token}" end + + test "uses default table names for the synthetic single-tenant backend", %{backend: backend} do + backend = %{backend | id: 0, token: nil} + + assert ClickHouseAdaptor.clickhouse_ingest_table_name(backend, :log) == + "otel_logs_default" + + assert ClickHouseAdaptor.clickhouse_ingest_table_name(backend, :metric) == + "otel_metrics_default" + + assert ClickHouseAdaptor.clickhouse_ingest_table_name(backend, :trace) == + "otel_traces_default" + end + + test "uses the configured suffix for the synthetic single-tenant backend", %{ + backend: backend + } do + Application.put_env(:logflare, :clickhouse_table_suffix, "custom") + backend = %{backend | id: 0, token: nil} + + assert ClickHouseAdaptor.clickhouse_ingest_table_name(backend, :log) == + "otel_logs_custom" + + assert ClickHouseAdaptor.clickhouse_ingest_table_name(backend, :metric) == + "otel_metrics_custom" + + assert ClickHouseAdaptor.clickhouse_ingest_table_name(backend, :trace) == + "otel_traces_custom" + end + + test "does not use the configured suffix outside single-tenant ClickHouse mode", %{ + backend: backend, + stringified_backend_token: token + } do + backend = %{backend | id: 0} + + for {single_tenant, backend_type} <- [ + {false, :clickhouse}, + {true, :postgres}, + {true, :bigquery} + ] do + Application.put_env(:logflare, :single_tenant, single_tenant) + Application.put_env(:logflare, :single_tenant_backend, backend_type) + + assert ClickHouseAdaptor.clickhouse_ingest_table_name(backend, :log) == + "otel_logs_#{token}" + end + end end describe "connection and basic functionality" do @@ -66,8 +124,6 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do {source, backend} = setup_clickhouse_test() - start_supervised!({ClickHouseAdaptor, backend}) - [source: source, backend: backend] end @@ -86,6 +142,17 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do assert is_integer(bytes) end + test "can execute queries for the single tenant backend", %{backend: backend} do + backend = %{backend | id: 0} + refute Backends.Cache.get_backend(backend.id) + Logflare.ContextCache.update(Backends, :get_backend, [backend.id], backend) + + assert {:ok, {[%{"test" => 1}], bytes}} = + ClickHouseAdaptor.execute_ch_query(backend, "SELECT 1 as test") + + assert is_integer(bytes) + end + test "handles query errors", %{backend: backend} do result = ClickHouseAdaptor.execute_ch_query(backend, "INVALID SQL QUERY") @@ -140,7 +207,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do test "describes a real ClickHouse NOT_AN_AGGREGATE error without the physical table", %{ backend: backend } do - assert :ok = ClickHouseAdaptor.provision_ingest_tables(backend) + provision_clickhouse_tables!(backend) table_name = ClickHouseAdaptor.clickhouse_ingest_table_name(backend, :log) assert {:error, @@ -1156,20 +1223,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do setup do insert(:plan, name: "Free") - {source, backend} = - setup_clickhouse_test( - config: %{ - read_only_urls: %{ - "api" => "http://localhost:8123", - "dashboard_logs" => "http://localhost:8123" - }, - default_read_cluster: "dashboard_logs" - } - ) - - start_supervised!({ClickHouseAdaptor, backend}) - - [source: source, backend: backend] + [backend: setup_read_cluster_backend()] end test "routes to the labeled pool for a configured read cluster", %{backend: backend} do @@ -1215,8 +1269,6 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do {source, backend} = setup_clickhouse_test() - start_supervised!({ClickHouseAdaptor, backend}) - [source: source, backend: backend] end @@ -1242,18 +1294,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do end test "falls back to the default cluster when the requested cluster is unhealthy" do - {_source, backend} = - setup_clickhouse_test( - config: %{ - read_only_urls: %{ - "api" => "http://localhost:8123", - "dashboard_logs" => "http://localhost:8123" - }, - default_read_cluster: "dashboard_logs" - } - ) - - start_supervised!({ClickHouseAdaptor, backend}) + backend = setup_read_cluster_backend() stub_read_cluster_connection_error(backend, "api") log = @@ -1273,18 +1314,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do end test "does not emit query error telemetry when the failover retry succeeds" do - {_source, backend} = - setup_clickhouse_test( - config: %{ - read_only_urls: %{ - "api" => "http://localhost:8123", - "dashboard_logs" => "http://localhost:8123" - }, - default_read_cluster: "dashboard_logs" - } - ) - - start_supervised!({ClickHouseAdaptor, backend}) + backend = setup_read_cluster_backend() stub_read_cluster_connection_error(backend, "api") TestUtils.attach_forwarder([:logflare, :clickhouse, :read_pool, :query_error]) TestUtils.attach_forwarder([:logflare, :clickhouse, :read_pool, :failover]) @@ -1306,18 +1336,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do end test "emits a single query error when the failover retry also fails" do - {_source, backend} = - setup_clickhouse_test( - config: %{ - read_only_urls: %{ - "api" => "http://localhost:8123", - "dashboard_logs" => "http://localhost:8123" - }, - default_read_cluster: "dashboard_logs" - } - ) - - start_supervised!({ClickHouseAdaptor, backend}) + backend = setup_read_cluster_backend() stub_read_cluster_connection_error(backend, ["api", "dashboard_logs"]) TestUtils.attach_forwarder([:logflare, :clickhouse, :read_pool, :query_error]) @@ -1339,18 +1358,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do end test "preserves the endpoint limit when falling back to the default cluster" do - {_source, backend} = - setup_clickhouse_test( - config: %{ - read_only_urls: %{ - "api" => "http://localhost:8123", - "dashboard_logs" => "http://localhost:8123" - }, - default_read_cluster: "dashboard_logs" - } - ) - - start_supervised!({ClickHouseAdaptor, backend}) + backend = setup_read_cluster_backend() stub_read_cluster_connection_error(backend, "api", self()) assert {:ok, %QueryResult{rows: rows}} = @@ -1383,7 +1391,6 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do } ) - start_supervised!({ClickHouseAdaptor, backend}) stub_read_cluster_queue_timeout(backend, "api_free") assert {:error, %QueryError{kind: :pool_exhausted}} = @@ -1403,7 +1410,6 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do } ) - start_supervised!({ClickHouseAdaptor, backend}) stub_read_cluster_connection_error(backend, "dashboard_logs") assert {:error, %QueryError{}} = @@ -1413,18 +1419,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do end test "falls back to the default cluster when the requested cluster pool cannot start" do - {_source, backend} = - setup_clickhouse_test( - config: %{ - read_only_urls: %{ - "api" => "http://localhost:8123", - "dashboard_logs" => "http://localhost:8123" - }, - default_read_cluster: "dashboard_logs" - } - ) - - start_supervised!({ClickHouseAdaptor, backend}) + backend = setup_read_cluster_backend() stub_read_cluster_pool_start_error(backend, "api") TestUtils.attach_forwarder([:logflare, :clickhouse, :read_pool, :query_error]) TestUtils.attach_forwarder([:logflare, :clickhouse, :read_pool, :failover]) @@ -1455,7 +1450,6 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do } ) - start_supervised!({ClickHouseAdaptor, backend}) stub_read_cluster_pool_start_error(backend, "dashboard_logs") TestUtils.attach_forwarder([:logflare, :clickhouse, :read_pool, :query_error]) @@ -1484,7 +1478,6 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do } ) - start_supervised!({ClickHouseAdaptor, backend}) stub_read_cluster_connection_error(backend, "adhoc") log = @@ -1511,7 +1504,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do end test "fails when the ingest cluster returns a connection error and logs the ingest URL" do - {_source, backend} = setup_clickhouse_test(cleanup?: false) + {_source, backend} = setup_clickhouse_test() stub(Ch, :query, fn _pool, _statement, _params, _opts -> {:error, %DBConnection.ConnectionError{message: "unreachable"}} @@ -1530,8 +1523,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do test "fails when the read cluster returns a connection error" do {_source, backend} = setup_clickhouse_test( - config: %{read_only_urls: %{"reporting" => "http://localhost:8123"}}, - cleanup?: false + config: %{read_only_urls: %{"reporting" => "http://localhost:8123"}} ) read_grant_statement = QueryTemplates.read_grant_check_statement() @@ -1559,7 +1551,6 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do } ) - start_supervised!({ClickHouseAdaptor, backend}) assert :ok = ClickHouseAdaptor.test_connection(backend) end @@ -1571,8 +1562,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do "reporting" => "http://localhost:8123", "adhoc" => "http://localhost:8124" } - }, - cleanup?: false + } ) stub(Ch, :start_link, fn opts -> @@ -1603,7 +1593,6 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do } ) - start_supervised!({ClickHouseAdaptor, backend}) assert :ok = ClickHouseAdaptor.test_connection(backend) end @@ -1616,7 +1605,6 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do } ) - start_supervised!({ClickHouseAdaptor, backend}) assert {:error, _} = ClickHouseAdaptor.test_connection(backend) end @@ -1629,7 +1617,6 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do } ) - start_supervised!({ClickHouseAdaptor, backend}) assert :ok = ClickHouseAdaptor.test_connection(backend) end end @@ -1672,8 +1659,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do read_only_urls: %{"reporting" => "http://localhost:8123"}, query_user: "ch_reader", query_password: "reader_pa55" - }, - cleanup?: false + } ) test_pid = self() @@ -1691,10 +1677,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do test "validates the query credentials against the primary url when no read cluster is set" do {_source, backend} = - setup_clickhouse_test( - config: %{query_user: "logflare", query_password: "logflare"}, - cleanup?: false - ) + setup_clickhouse_test(config: %{query_user: "logflare", query_password: "logflare"}) read_grant_statement = QueryTemplates.read_grant_check_statement() test_pid = self() @@ -1710,7 +1693,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do end test "skips the read grant check when neither a query user nor a read cluster is set" do - {_source, backend} = setup_clickhouse_test(cleanup?: false) + {_source, backend} = setup_clickhouse_test() read_grant_statement = QueryTemplates.read_grant_check_statement() test_pid = self() @@ -1727,10 +1710,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do test "fails when the dedicated query user cannot authenticate" do {_source, backend} = - setup_clickhouse_test( - config: %{query_user: "ch_reader", query_password: "reader_pa55"}, - cleanup?: false - ) + setup_clickhouse_test(config: %{query_user: "ch_reader", query_password: "reader_pa55"}) log = ExUnit.CaptureLog.capture_log(fn -> @@ -1748,8 +1728,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do config: %{ read_only_urls: %{"reporting" => "http://localhost:8123"}, query_user: "ch_reader" - }, - cleanup?: false + } ) test_pid = self() @@ -1773,10 +1752,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do test "succeeds when only the read grant check fails" do {_source, backend} = - setup_clickhouse_test( - config: %{query_user: "ch_reader", query_password: "reader_pa55"}, - cleanup?: false - ) + setup_clickhouse_test(config: %{query_user: "ch_reader", query_password: "reader_pa55"}) log = ExUnit.CaptureLog.capture_log(fn -> @@ -1788,7 +1764,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do end test "fails when the ingest grant check fails" do - {_source, backend} = setup_clickhouse_test(cleanup?: false) + {_source, backend} = setup_clickhouse_test() grant_check_statement = QueryTemplates.grant_check_statement() @@ -2007,8 +1983,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do {source, backend} = setup_clickhouse_test() - start_supervised!({ClickHouseAdaptor, backend}) - assert :ok = ClickHouseAdaptor.provision_ingest_tables(backend) + provision_clickhouse_tables!(backend) [source: source, backend: backend] end @@ -2222,8 +2197,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do {source, backend} = setup_clickhouse_test() - start_supervised!({ClickHouseAdaptor, backend}) - assert :ok = ClickHouseAdaptor.provision_ingest_tables(backend) + provision_clickhouse_tables!(backend) [source: source, backend: backend] end @@ -2284,8 +2258,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do {source, backend} = setup_clickhouse_test() - start_supervised!({ClickHouseAdaptor, backend}) - assert :ok = ClickHouseAdaptor.provision_ingest_tables(backend) + provision_clickhouse_tables!(backend) [source: source, backend: backend] end @@ -2334,36 +2307,16 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do end end - describe "execute_query/2" do + describe "execute_query/2 replica routing" do setup do insert(:plan, name: "Free") - {source, backend} = setup_clickhouse_test() - - start_supervised!({ClickHouseAdaptor, backend}) + {_source, backend} = setup_clickhouse_test(config: %{replica_routing_param: "project"}) - [source: source, backend: backend] - end - - test "executes simple queries using backend-only interface", %{backend: backend} do - result = ClickHouseAdaptor.execute_query(backend, "SELECT 1 as test_value", []) - - assert {:ok, %QueryResult{rows: [%{"test_value" => 1}]}} = result - end - - test "converts `@param` syntax to ClickHouse `{param:String}` format", %{backend: backend} do - result = - ClickHouseAdaptor.execute_query( - backend, - {"SELECT @test_value as param_result", ["test_value"], %{"test_value" => "hello"}}, - [] - ) - - assert {:ok, %QueryResult{rows: [%{"param_result" => "hello"}]}} = result + [backend: backend] end - test "sends the configured routing param as the replica tag header" do - backend = start_replica_routing_backend("project") + test "sends the configured routing param as the replica tag header", %{backend: backend} do parent = self() expect(Ch, :query, fn pool, statement, params, opts -> @@ -2381,8 +2334,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do assert_received {:ch_headers, [{"x-clickhouse-replica-tag", "abcdefghij"}]} end - test "sends the replica tag header on the max_limit endpoint path" do - backend = start_replica_routing_backend("project") + test "sends the replica tag header on the max_limit endpoint path", %{backend: backend} do parent = self() expect(Ch, :query, fn pool, statement, params, opts -> @@ -2400,8 +2352,8 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do assert_received {:ch_headers, [{"x-clickhouse-replica-tag", "abcdefghij"}]} end - test "does not send a replica tag header when the routing param is absent from the request" do - backend = start_replica_routing_backend("project") + test "does not send a replica tag header when the routing param is absent from the request", + %{backend: backend} do parent = self() expect(Ch, :query, fn pool, statement, params, opts -> @@ -2419,7 +2371,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do assert_received {:ch_headers, []} end - test "does not send a replica tag header when the backend has no routing param configured", + test "does not send a replica tag header for raw string queries without request params", %{backend: backend} do parent = self() @@ -2429,17 +2381,12 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do end) assert {:ok, %QueryResult{}} = - ClickHouseAdaptor.execute_query( - backend, - {"SELECT 1 as test", [], %{"project" => "abcdefghij"}}, - [] - ) + ClickHouseAdaptor.execute_query(backend, "SELECT 1 as test", []) assert_received {:ch_headers, []} end - test "does not send a replica tag header for raw string queries without request params" do - backend = start_replica_routing_backend("project") + test "ignores a routing param value that is not a printable token", %{backend: backend} do parent = self() expect(Ch, :query, fn pool, statement, params, opts -> @@ -2448,13 +2395,16 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do end) assert {:ok, %QueryResult{}} = - ClickHouseAdaptor.execute_query(backend, "SELECT 1 as test", []) + ClickHouseAdaptor.execute_query( + backend, + {"SELECT 1 as test", [], %{"project" => "abc\r\nX-Injected: 1"}}, + [] + ) assert_received {:ch_headers, []} end - test "ignores a routing param value that is not a printable token" do - backend = start_replica_routing_backend("project") + test "ignores a routing param value that is not a string", %{backend: backend} do parent = self() expect(Ch, :query, fn pool, statement, params, opts -> @@ -2465,15 +2415,14 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do assert {:ok, %QueryResult{}} = ClickHouseAdaptor.execute_query( backend, - {"SELECT 1 as test", [], %{"project" => "abc\r\nX-Injected: 1"}}, + {"SELECT 1 as test", [], %{"project" => ["a", "b"]}}, [] ) assert_received {:ch_headers, []} end - test "ignores a routing param value that is not a string" do - backend = start_replica_routing_backend("project") + test "ignores a routing param value containing non-ASCII characters", %{backend: backend} do parent = self() expect(Ch, :query, fn pool, statement, params, opts -> @@ -2484,16 +2433,16 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do assert {:ok, %QueryResult{}} = ClickHouseAdaptor.execute_query( backend, - {"SELECT 1 as test", [], %{"project" => ["a", "b"]}}, + {"SELECT 1 as test", [], %{"project" => "café"}}, [] ) assert_received {:ch_headers, []} end - test "ignores a routing param value containing non-ASCII characters" do - backend = start_replica_routing_backend("project") + test "sends a routing param value of exactly 256 bytes", %{backend: backend} do parent = self() + value = String.duplicate("a", 256) expect(Ch, :query, fn pool, statement, params, opts -> send(parent, {:ch_headers, Keyword.get(opts, :headers, [])}) @@ -2503,17 +2452,15 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do assert {:ok, %QueryResult{}} = ClickHouseAdaptor.execute_query( backend, - {"SELECT 1 as test", [], %{"project" => "café"}}, + {"SELECT 1 as test", [], %{"project" => value}}, [] ) - assert_received {:ch_headers, []} + assert_received {:ch_headers, [{"x-clickhouse-replica-tag", ^value}]} end - test "sends a routing param value of exactly 256 bytes" do - backend = start_replica_routing_backend("project") + test "ignores a routing param value of 257 bytes", %{backend: backend} do parent = self() - value = String.duplicate("a", 256) expect(Ch, :query, fn pool, statement, params, opts -> send(parent, {:ch_headers, Keyword.get(opts, :headers, [])}) @@ -2523,15 +2470,42 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do assert {:ok, %QueryResult{}} = ClickHouseAdaptor.execute_query( backend, - {"SELECT 1 as test", [], %{"project" => value}}, + {"SELECT 1 as test", [], %{"project" => String.duplicate("a", 257)}}, [] ) - assert_received {:ch_headers, [{"x-clickhouse-replica-tag", ^value}]} + assert_received {:ch_headers, []} + end + end + + describe "execute_query/2" do + setup do + insert(:plan, name: "Free") + + {source, backend} = setup_clickhouse_test() + + [source: source, backend: backend] + end + + test "executes simple queries using backend-only interface", %{backend: backend} do + result = ClickHouseAdaptor.execute_query(backend, "SELECT 1 as test_value", []) + + assert {:ok, %QueryResult{rows: [%{"test_value" => 1}]}} = result end - test "ignores a routing param value of 257 bytes" do - backend = start_replica_routing_backend("project") + test "converts `@param` syntax to ClickHouse `{param:String}` format", %{backend: backend} do + result = + ClickHouseAdaptor.execute_query( + backend, + {"SELECT @test_value as param_result", ["test_value"], %{"test_value" => "hello"}}, + [] + ) + + assert {:ok, %QueryResult{rows: [%{"param_result" => "hello"}]}} = result + end + + test "does not send a replica tag header when the backend has no routing param configured", + %{backend: backend} do parent = self() expect(Ch, :query, fn pool, statement, params, opts -> @@ -2542,7 +2516,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do assert {:ok, %QueryResult{}} = ClickHouseAdaptor.execute_query( backend, - {"SELECT 1 as test", [], %{"project" => String.duplicate("a", 257)}}, + {"SELECT 1 as test", [], %{"project" => "abcdefghij"}}, [] ) @@ -2550,19 +2524,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do end test "keeps the replica tag header on the default cluster failover retry" do - {_source, backend} = - setup_clickhouse_test( - config: %{ - replica_routing_param: "project", - read_only_urls: %{ - "api" => "http://localhost:8123", - "dashboard_logs" => "http://localhost:8123" - }, - default_read_cluster: "dashboard_logs" - } - ) - - start_supervised!({ClickHouseAdaptor, backend}, id: {:replica_routing, backend.id}) + backend = setup_read_cluster_backend(%{replica_routing_param: "project"}) parent = self() backend_id = backend.id @@ -3030,7 +2992,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do source: source, backend: backend } do - assert :ok = ClickHouseAdaptor.provision_ingest_tables(backend) + provision_clickhouse_tables!(backend) log_events = for i <- 1..5 do @@ -3096,8 +3058,6 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do default_ingest?: true ) - start_supervised!({ClickHouseAdaptor, backend}) - [backend: backend, user: user, source1: source1_with_backend, source2: source2] end @@ -3209,7 +3169,6 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do insert(:plan, name: "Free") {source, backend} = setup_clickhouse_test() - start_supervised!({ClickHouseAdaptor, backend}) [source: source, backend: backend] end @@ -3258,7 +3217,6 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do backend: backend1 } do {_source2, backend2} = setup_clickhouse_test(source: source) - start_supervised!({ClickHouseAdaptor, backend2}, id: :adaptor2) {:ok, _} = ClickHouseAdaptor.execute_ch_query(backend2, "SELECT 1") assert ConnectionManager.pool_active?(backend2) @@ -3276,7 +3234,6 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do insert(:plan, name: "Free") {source, backend} = setup_clickhouse_test() - start_supervised!({ClickHouseAdaptor, backend}) [source: source, backend: backend] end @@ -3485,9 +3442,21 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptorTest do {query, declared_params, input_params, %EndpointQuery{max_limit: max_limit}} end - defp start_replica_routing_backend(param) do - {_source, backend} = setup_clickhouse_test(config: %{replica_routing_param: param}) - start_supervised!({ClickHouseAdaptor, backend}, id: {:replica_routing, backend.id}) + @spec setup_read_cluster_backend(map()) :: Backend.t() + defp setup_read_cluster_backend(config \\ %{}) do + config = + Map.merge( + %{ + read_only_urls: %{ + "api" => "http://localhost:8123", + "dashboard_logs" => "http://localhost:8123" + }, + default_read_cluster: "dashboard_logs" + }, + config + ) + + {_source, backend} = setup_clickhouse_test(config: config) backend end diff --git a/test/logflare/backends/bigquery_adaptor_test.exs b/test/logflare/backends/bigquery_adaptor_test.exs index 802b7f84ee..a95c1812c5 100644 --- a/test/logflare/backends/bigquery_adaptor_test.exs +++ b/test/logflare/backends/bigquery_adaptor_test.exs @@ -10,6 +10,8 @@ defmodule Logflare.Backends.BigQueryAdaptorTest do alias Logflare.Backends.SourceSup alias Logflare.Backends.Adaptor.BigQueryAdaptor alias Logflare.Backends.Adaptor.QueryResult + alias Logflare.Google.CloudResourceManager + alias Logflare.SingleTenant alias Logflare.SystemMetrics.AllLogsLogged alias GoogleApi.CloudResourceManager.V1.Model @@ -498,6 +500,37 @@ defmodule Logflare.Backends.BigQueryAdaptorTest do assert_receive {:created, "logflare-managed-1"}, 1000 end + test "on_system_start/0 provisions missing accounts and updates IAM policy" do + expect(GoogleApi.IAM.V1.Api.Projects, :iam_projects_service_accounts_list, fn + _conn, "projects/" <> project_id, _opts -> + {:ok, + %{ + accounts: [ + %GoogleApi.IAM.V1.Model.ServiceAccount{ + email: "logflare-managed-0@#{project_id}.iam.gserviceaccount.com", + name: + "projects/#{project_id}/serviceAccounts/logflare-managed-0@#{project_id}.iam.gserviceaccount.com" + } + ], + nextPageToken: nil + }} + end) + + expect(GoogleApi.IAM.V1.Api.Projects, :iam_projects_service_accounts_create, fn + _conn, "projects/" <> project_id, opts -> + assert %{accountId: "logflare-managed-1"} = opts[:body] + + {:ok, + %GoogleApi.IAM.V1.Model.ServiceAccount{ + email: "logflare-managed-1@#{project_id}.iam.gserviceaccount.com" + }} + end) + + expect(CloudResourceManager, :set_iam_policy, fn [async: false] -> :ok end) + + assert :ok = BigQueryAdaptor.on_system_start() + end + test "update_iam_policy/0" do pid = self() # Mock IAM API calls for listing existing service accounts @@ -537,6 +570,12 @@ defmodule Logflare.Backends.BigQueryAdaptorTest do end end + test "on_supabase_start/0 updates source schemas" do + expect(SingleTenant, :update_supabase_source_schemas, fn -> :ok end) + + assert :ok = BigQueryAdaptor.on_supabase_start() + end + describe "managed service accounts is disabled" do setup do original_pool_size = diff --git a/test/logflare/backends/consolidated_sup_test.exs b/test/logflare/backends/consolidated_sup_test.exs index ef011c1e6a..94fd056192 100644 --- a/test/logflare/backends/consolidated_sup_test.exs +++ b/test/logflare/backends/consolidated_sup_test.exs @@ -10,6 +10,7 @@ defmodule Logflare.Backends.ConsolidatedSupTest do insert(:plan, name: "Free") {_source, backend} = setup_clickhouse_test() + on_exit(fn -> ConsolidatedSup.stop_pipeline(backend.id) end) [backend: backend] end @@ -82,14 +83,6 @@ defmodule Logflare.Backends.ConsolidatedSupTest do IngestEventQueue.upsert_tid({:consolidated, backend.id, nil}) - on_exit(fn -> - try do - ConsolidatedSup.stop_pipeline(backend.id) - catch - _kind, _value -> :ok - end - end) - [source: source, backend: backend] end @@ -97,7 +90,7 @@ defmodule Logflare.Backends.ConsolidatedSupTest do source: source, backend: backend } do - assert :ok = ClickHouseAdaptor.provision_ingest_tables(backend) + provision_clickhouse_tables!(backend) assert {:ok, _pid} = ConsolidatedSup.start_pipeline(backend) events = for _ <- 1..5, do: build(:log_event, source: source) diff --git a/test/logflare/backends/consolidated_sup_worker_test.exs b/test/logflare/backends/consolidated_sup_worker_test.exs index cd3ae8ffe6..6966b7a596 100644 --- a/test/logflare/backends/consolidated_sup_worker_test.exs +++ b/test/logflare/backends/consolidated_sup_worker_test.exs @@ -1,8 +1,10 @@ defmodule Logflare.Backends.ConsolidatedSupWorkerTest do use Logflare.DataCase, async: false + alias Logflare.Backends alias Logflare.Backends.ConsolidatedSup alias Logflare.Backends.ConsolidatedSupWorker + alias Logflare.SingleTenant describe "ConsolidatedSupWorker" do setup do @@ -77,4 +79,26 @@ defmodule Logflare.Backends.ConsolidatedSupWorkerTest do refute ConsolidatedSup.pipeline_running?(backend.id) end end + + describe "ConsolidatedSupWorker single-tenant default" do + TestUtils.setup_single_tenant(seed_user: true, backend_type: :clickhouse) + + test "starts the synthetic consolidated backend without sources or rules" do + backend = SingleTenant.get_default_clickhouse_backend() + assert backend.id == 0 + + on_exit(fn -> ConsolidatedSup.stop_pipeline(backend.id) end) + + refute ConsolidatedSup.pipeline_running?(backend.id) + worker = start_supervised!({ConsolidatedSupWorker, [interval: 10_000]}) + + send(worker, :check) + + TestUtils.retry_assert(fn -> + assert ConsolidatedSup.pipeline_running?(backend.id) + end) + + assert Backends.list_backends(user_id: backend.user_id) == [] + end + end end diff --git a/test/logflare/backends/ingest_event_queue_test.exs b/test/logflare/backends/ingest_event_queue_test.exs index 5bdd37039f..58669a925f 100644 --- a/test/logflare/backends/ingest_event_queue_test.exs +++ b/test/logflare/backends/ingest_event_queue_test.exs @@ -1804,7 +1804,7 @@ defmodule Logflare.Backends.IngestEventQueueTest do end end - test "BufferCacheWorker caches buffer lengths every n seconds" do + test "BufferCacheWorker caches source and consolidated buffer lengths every n seconds" do user = insert(:user) source = insert(:source, user: user) backend = insert(:backend, user: user) @@ -1812,19 +1812,27 @@ defmodule Logflare.Backends.IngestEventQueueTest do table = {source.id, backend.id, pid} other_table = {source.id, nil, pid} + consolidated_table = {:consolidated, backend.id, pid} IngestEventQueue.upsert_tid(table) IngestEventQueue.upsert_tid(other_table) + IngestEventQueue.upsert_tid(consolidated_table) start_supervised!({BufferCacheWorker, interval: 100}) le = build(:log_event, source: source) IngestEventQueue.add_to_table(table, [le]) IngestEventQueue.add_to_table(other_table, [le]) + IngestEventQueue.add_to_table(consolidated_table, [le]) + node = Node.self() # Verify worker cached the values automatically (without manual cache calls) TestUtils.retry_assert(fn -> assert PubSubRates.Cache.get_cluster_buffers(source.id, backend.id) == 1 assert PubSubRates.Cache.get_cluster_buffers(source.id, nil) == 1 + assert PubSubRates.Cache.get_cluster_buffers(:consolidated, backend.id) == 1 + + assert {:ok, %{^node => %{len: 1}}} = + Cachex.get(PubSubRates.Cache, {:consolidated, backend.id, "buffers"}) end) end diff --git a/test/logflare/backends/spool/memory_monitor_test.exs b/test/logflare/backends/spool/memory_monitor_test.exs index 7c0bf2167f..1e1f76d117 100644 --- a/test/logflare/backends/spool/memory_monitor_test.exs +++ b/test/logflare/backends/spool/memory_monitor_test.exs @@ -223,7 +223,11 @@ defmodule Logflare.Backends.Spool.MemoryMonitorTest do # completely backlogged and it would report "not full" regardless. backend = insert(:backend, user: user, type: :clickhouse, default_ingest?: false) {:ok, _source} = Backends.update_source_backends(source, [backend]) - backend_table_key = {source.id, backend.id, self()} + + {queue_owner, backend_id} = + Backends.ingest_queue_key(source, Backends.get_backend(backend.id)) + + backend_table_key = {queue_owner, backend_id, self()} IngestEventQueue.upsert_tid(backend_table_key) fill_queue(backend_table_key) diff --git a/test/logflare/backends/system_backend_test.exs b/test/logflare/backends/system_backend_test.exs new file mode 100644 index 0000000000..90ce6a740b --- /dev/null +++ b/test/logflare/backends/system_backend_test.exs @@ -0,0 +1,44 @@ +defmodule Logflare.Backends.SystemBackendTest do + use Logflare.DataCase, async: false + + alias Logflare.Backends.Adaptor.BigQueryAdaptor + alias Logflare.Backends.SystemBackend + alias Logflare.Sources.Source + + TestUtils.setup_single_tenant(backend_type: :bigquery) + + test "dispatches lifecycle hooks to the single-tenant BigQuery adaptor" do + assert_bigquery_hooks_dispatched(build(:source)) + end + + test "multi-tenant deployments use BigQuery regardless of the single-tenant selection" do + Application.put_env(:logflare, :single_tenant, false) + Application.put_env(:logflare, :single_tenant_backend, :clickhouse) + assert_bigquery_hooks_dispatched(build(:source)) + end + + test "PostgreSQL and ClickHouse do not run BigQuery setup" do + reject(BigQueryAdaptor, :on_system_start, 0) + reject(BigQueryAdaptor, :on_source_start, 1) + reject(BigQueryAdaptor, :on_supabase_start, 0) + + for type <- [:postgres, :clickhouse] do + Application.put_env(:logflare, :single_tenant_backend, type) + + assert :ok = SystemBackend.on_system_start() + assert :ok = SystemBackend.on_source_start(build(:source)) + assert :ok = SystemBackend.on_supabase_start() + end + end + + @spec assert_bigquery_hooks_dispatched(Source.t()) :: :ok + defp assert_bigquery_hooks_dispatched(source) do + expect(BigQueryAdaptor, :on_system_start, fn -> :ok end) + expect(BigQueryAdaptor, :on_source_start, fn ^source -> :ok end) + expect(BigQueryAdaptor, :on_supabase_start, fn -> :ok end) + + assert :ok = SystemBackend.on_system_start() + assert :ok = SystemBackend.on_source_start(source) + assert :ok = SystemBackend.on_supabase_start() + end +end diff --git a/test/logflare/backends_test.exs b/test/logflare/backends_test.exs index c11960098a..8271210a14 100644 --- a/test/logflare/backends_test.exs +++ b/test/logflare/backends_test.exs @@ -1396,6 +1396,22 @@ defmodule Logflare.BackendsTest do end) end + test "starts a non-shared default backend for the source", %{ + source: source, + user: user + } do + default_backend = insert(:backend, type: :webhook, user: user) + default_backend = Backends.get_backend(default_backend.id) + + stub(Backends, :get_default_backend, fn %User{id: user_id} when user_id == user.id -> + default_backend + end) + + start_supervised!({SourceSup, source}) + + assert SourceSup.backend_child_started?(default_backend.id, source.id) + end + test "SourceSup filters out consolidated backends", %{source: source, user: user} do consolidated_backend = insert(:backend, @@ -1540,7 +1556,7 @@ defmodule Logflare.BackendsTest do test "any_ingest_queue_over_limit?/1 is true for a backend that is not flagged default_ingest?", %{source: source} do user = Repo.get(User, source.user_id) - backend = insert(:backend, user: user, type: :clickhouse, default_ingest?: false) + backend = insert(:backend, user: user, type: :postgres, default_ingest?: false) {:ok, _} = Backends.update_source_backends(source, [backend]) table_key = {source.id, backend.id, self()} @@ -1607,6 +1623,60 @@ defmodule Logflare.BackendsTest do end end + describe "single-tenant ClickHouse default backend" do + TestUtils.setup_single_tenant(backend_type: :clickhouse) + + setup do + insert(:plan) + user = insert(:user) + source = insert(:source, user: user) + backend = Backends.get_default_backend(user) + + [backend: backend, source: source] + end + + test "does not start ingestion separately for each source", %{ + backend: backend, + source: source + } do + start_supervised!({SourceSup, source}) + + refute SourceSup.backend_child_started?(backend.id, source.id) + end + + test "sends events only to shared queue", %{ + backend: backend, + source: source + } do + {:ok, _} = Registry.register(Logflare.Backends.SourceRegistry, {source.id, SourceSup}, nil) + IngestEventQueue.upsert_tid({:consolidated, backend.id, nil}) + IngestEventQueue.upsert_tid({source.id, nil, nil}) + + assert {:ok, 1} = Backends.ingest_logs([build(:log_event, source: source)], source) + + assert {:ok, [_]} = IngestEventQueue.fetch_events({:consolidated, backend.id}, 10) + + assert {:ok, []} = IngestEventQueue.fetch_events({source.id, nil}, 10) + end + + test "checks only shared queue for overflow", %{ + backend: backend, + source: source + } do + legacy_table_key = {source.id, nil, self()} + IngestEventQueue.upsert_tid(legacy_table_key) + fill_queue_over_limit(legacy_table_key) + + refute Backends.any_ingest_queue_over_limit?(source.id) + + consolidated_table_key = {:consolidated, backend.id, self()} + IngestEventQueue.upsert_tid(consolidated_table_key) + fill_queue_over_limit(consolidated_table_key) + + assert Backends.any_ingest_queue_over_limit?(source.id) + end + end + describe "ingest filters" do setup do insert(:plan) diff --git a/test/logflare/endpoints_test.exs b/test/logflare/endpoints_test.exs index f223e7e84a..f14ad37449 100644 --- a/test/logflare/endpoints_test.exs +++ b/test/logflare/endpoints_test.exs @@ -832,8 +832,7 @@ defmodule Logflare.EndpointsTest do source = insert(:source, user: user, name: "c") {_source, backend} = setup_clickhouse_test(user: user, source: source) - start_supervised!({ClickHouseAdaptor, backend}) - assert :ok = ClickHouseAdaptor.provision_ingest_tables(backend) + provision_clickhouse_tables!(backend) log_events = for i <- 1..5 do diff --git a/test/logflare/networking_test.exs b/test/logflare/networking_test.exs index 0a7fc63ad2..86d025c0e1 100644 --- a/test/logflare/networking_test.exs +++ b/test/logflare/networking_test.exs @@ -5,6 +5,20 @@ defmodule Logflare.NetworkingTest do alias Logflare.Backends.Adaptor.DatadogAdaptor alias Logflare.Networking + describe "multi-tenant mode" do + setup :use_non_test_networking_config + + test "returns BigQuery, gRPC, and ClickHouse connection pools" do + assert Logflare.FinchGoth in finch_names() + assert Logflare.FinchIngest in finch_names() + assert Logflare.FinchQuery in finch_names() + assert Logflare.FinchClickHouseIngest in finch_names() + + assert {Logflare.Networking.GrpcPool, _opts} = + Enum.find(Networking.pools(), &match?({Logflare.Networking.GrpcPool, _}, &1)) + end + end + describe "single tenant mode using Big Query" do TestUtils.setup_single_tenant() @@ -16,22 +30,14 @@ defmodule Logflare.NetworkingTest do end test "returns bigquery, clickhouse, and spool connection pools" do - finch_names = - Networking.pools() - |> Enum.filter(fn - {mod, _} -> mod == Finch - _ -> false - end) - |> Enum.map(fn {Finch, opts} -> Keyword.get(opts, :name) end) - - assert finch_names == [ + assert finch_names() == [ Logflare.FinchGoth, - Logflare.FinchDefaultHttp1, Logflare.FinchIngest, Logflare.FinchQuery, Logflare.FinchDefault, Logflare.FinchSpoolS3, Logflare.FinchSpoolSQS, + Logflare.FinchDefaultHttp1, Logflare.FinchClickHouseIngest, Logflare.FinchClickHouseAsyncIngest, Logflare.FinchS3 @@ -58,14 +64,7 @@ defmodule Logflare.NetworkingTest do |> Map.put(:default, protocols: [:http1]) assert [ - {Finch, - name: Logflare.FinchDefaultHttp1, - pools: %{default: [protocols: [:http1], size: 50]}}, - {Finch, - [ - name: Logflare.FinchDefault, - pools: datadog_pools - ]}, + {Finch, [name: Logflare.FinchDefault, pools: datadog_pools]}, {Finch, name: Logflare.FinchSpoolS3, pools: %{ @@ -76,6 +75,11 @@ defmodule Logflare.NetworkingTest do pools: %{ default: _spool_sqs_config }}, + {Finch, + [ + name: Logflare.FinchDefaultHttp1, + pools: %{default: [protocols: [:http1], size: 50]} + ]}, {Finch, name: Logflare.FinchClickHouseIngest, pools: %{ @@ -193,4 +197,27 @@ defmodule Logflare.NetworkingTest do end end end + + describe "single tenant mode using ClickHouse" do + TestUtils.setup_single_tenant(backend_type: :clickhouse) + setup :use_non_test_networking_config + + test "excludes BigQuery and gRPC connection pools" do + assert finch_names() == [ + Logflare.FinchDefault, + Logflare.FinchDefaultHttp1, + Logflare.FinchClickHouseIngest, + Logflare.FinchClickHouseAsyncIngest, + Logflare.FinchS3 + ] + + refute Enum.any?(Networking.pools(), &match?({Logflare.Networking.GrpcPool, _}, &1)) + end + end + + defp use_non_test_networking_config(_context) do + previous_env = Application.get_env(:logflare, :env) + Application.put_env(:logflare, :env, :dev) + on_exit(fn -> Application.put_env(:logflare, :env, previous_env) end) + end end diff --git a/test/logflare/single_tenant_test.exs b/test/logflare/single_tenant_test.exs index 1605762789..a8118f050b 100644 --- a/test/logflare/single_tenant_test.exs +++ b/test/logflare/single_tenant_test.exs @@ -2,19 +2,53 @@ defmodule Logflare.SingleTenantTest do @moduledoc false use Logflare.DataCase import Logflare.Utils.Guards - alias Logflare.SingleTenant + alias Logflare.Auth + alias Logflare.Backends + alias Logflare.Backends.Adaptor.BigQueryAdaptor + alias Logflare.Backends.Backend + alias Logflare.Backends.ConsolidatedSup alias Logflare.Billing - alias Logflare.Users - alias Logflare.User alias Logflare.Billing.Plan - alias Logflare.Sources alias Logflare.Endpoints - alias Logflare.Sources.Source.BigQuery.Schema + alias Logflare.SingleTenant + alias Logflare.Sources alias Logflare.Sources.Source - alias Logflare.Auth - alias Logflare.Backends.Backend - alias Logflare.Backends - alias Logflare.Backends.Adaptor.BigQueryAdaptor + alias Logflare.Sources.Source.BigQuery.Schema + alias Logflare.User + alias Logflare.Users + + describe "ClickHouse connection URL" do + test "builds adapter options from a URL" do + for {url, expected} <- [ + { + "https://logflare%40example.com:s%3Acret@clickhouse.example.com:9440/otel%5Flogs", + [ + url: "https://clickhouse.example.com:9440", + username: "logflare@example.com", + password: "s:cret", + database: "otel_logs", + port: 9440 + ] + }, + { + "https://localhost/logs", + [url: "https://localhost", database: "logs", port: 8443] + }, + { + "http://localhost", + [url: "http://localhost", database: "default", port: 8123] + } + ] do + assert SingleTenant.clickhouse_backend_adapter_opts_from_url!(url) == expected + end + end + + test "requires a hostname" do + assert_raise RuntimeError, ~r/must include a hostname/, fn -> + SingleTenant.clickhouse_backend_adapter_opts_from_url!("not-a-url") + end + end + end describe "single tenant mode using Big Query" do TestUtils.setup_single_tenant() @@ -79,6 +113,8 @@ defmodule Logflare.SingleTenantTest do assert {:ok, user} = SingleTenant.create_default_user() assert %Backend{type: :bigquery} = SingleTenant.get_default_backend() assert %Backend{type: :bigquery} = Backends.get_default_backend(user) + assert SingleTenant.get_default_clickhouse_backend() == nil + assert SingleTenant.lookup_system_default_backend(%Source{user_id: user.id}) == nil end test "single_tenant? returns true when in single tenant mode" do @@ -86,13 +122,39 @@ defmodule Logflare.SingleTenantTest do end test "Logflare.Application.startup_tasks/0 should insert plan and user" do - expect(BigQueryAdaptor, :update_iam_policy, fn -> :ok end) + expect(BigQueryAdaptor, :on_system_start, fn -> + assert SingleTenant.get_default_user() == nil + :ok + end) Logflare.Application.startup_tasks() assert [_] = Billing.list_plans() assert 1 = Users.count_users() end + + test "startup invokes Supabase setup after sources and endpoints are seeded" do + previous_supabase_mode = Application.get_env(:logflare, :supabase_mode) + Application.put_env(:logflare, :supabase_mode, true) + on_exit(fn -> Application.put_env(:logflare, :supabase_mode, previous_supabase_mode) end) + + expect(BigQueryAdaptor, :on_system_start, fn -> :ok end) + expect(SingleTenant, :create_supabase_sources, fn -> send(self(), :sources_seeded) end) + expect(SingleTenant, :create_supabase_endpoints, fn -> send(self(), :endpoints_seeded) end) + + expect(SingleTenant, :ensure_supabase_sources_started, fn -> + send(self(), :sources_started) + end) + + expect(BigQueryAdaptor, :on_supabase_start, fn -> + assert_received :sources_seeded + assert_received :endpoints_seeded + assert_received :sources_started + :ok + end) + + Logflare.Application.startup_tasks() + end end describe ":postgres_backend_url" do @@ -105,10 +167,13 @@ defmodule Logflare.SingleTenantTest do url = "postgresql://#{username}:#{password}@#{hostname}/#{database}" prev = Application.get_env(:logflare, :postgres_backend_adapter) + previous_backend = Application.get_env(:logflare, :single_tenant_backend) Application.put_env(:logflare, :postgres_backend_adapter, url: url) + Application.put_env(:logflare, :single_tenant_backend, :postgres) on_exit(fn -> Application.put_env(:logflare, :postgres_backend_adapter, prev) + Application.put_env(:logflare, :single_tenant_backend, previous_backend) end) [url: url, password: password] @@ -138,6 +203,24 @@ defmodule Logflare.SingleTenantTest do test "single_tenant? returns false when not in single tenant mode" do refute SingleTenant.single_tenant?() + refute SingleTenant.bigquery_backend?() + assert Backends.bigquery_default_backend?() + end + + test "backend_type/0 requires runtime configuration" do + previous_backend = Application.fetch_env(:logflare, :single_tenant_backend) + Application.delete_env(:logflare, :single_tenant_backend) + + on_exit(fn -> + case previous_backend do + {:ok, backend} -> Application.put_env(:logflare, :single_tenant_backend, backend) + :error -> Application.delete_env(:logflare, :single_tenant_backend) + end + end) + + assert_raise ArgumentError, ~r/could not fetch application environment/, fn -> + SingleTenant.backend_type() + end end describe "supabase_mode=true using Big Query" do @@ -187,6 +270,14 @@ defmodule Logflare.SingleTenantTest do describe "single tenant mode using Postgres" do TestUtils.setup_single_tenant(backend_type: :postgres) + test "does not return a synthetic ClickHouse backend" do + assert {:ok, _plan} = SingleTenant.create_default_plan() + assert {:ok, user} = SingleTenant.create_default_user() + + assert SingleTenant.get_default_clickhouse_backend() == nil + assert SingleTenant.lookup_system_default_backend(%Source{user_id: user.id}) == nil + end + test "create_default_plan/0 creates default enterprise plan if not present" do assert {:ok, plan} = SingleTenant.create_default_plan() assert plan.name == "Enterprise" @@ -232,6 +323,70 @@ defmodule Logflare.SingleTenantTest do end end + describe "single tenant mode using ClickHouse" do + TestUtils.setup_single_tenant(backend_type: :clickhouse) + + test "returns nil before the default user exists" do + assert SingleTenant.get_default_user() == nil + assert SingleTenant.get_default_clickhouse_backend() == nil + assert Backends.get_backend(0) == nil + end + + test "rejects plain maps in place of user structs" do + assert_raise FunctionClauseError, fn -> + apply(SingleTenant, :get_default_clickhouse_backend, [%{id: 123}]) + end + end + + test "source lookup uses the source owner rather than the default user" do + assert {:ok, _plan} = SingleTenant.create_default_plan() + assert {:ok, default_user} = SingleTenant.create_default_user() + source = %Source{user_id: default_user.id + 1} + + assert %Backend{id: 0, token: nil, user_id: user_id, consolidated_ingest?: true} = + SingleTenant.lookup_system_default_backend(source) + + assert user_id == source.user_id + assert SingleTenant.get_default_clickhouse_backend().user_id == default_user.id + end + + test "startup uses a synthetic ClickHouse backend" do + expect(SingleTenant, :create_supabase_sources, fn -> {:ok, []} end) + expect(SingleTenant, :create_supabase_endpoints, fn -> {:ok, []} end) + expect(SingleTenant, :ensure_supabase_sources_started, fn -> :ok end) + + previous_supabase_mode = Application.get_env(:logflare, :supabase_mode) + Application.put_env(:logflare, :supabase_mode, true) + + on_exit(fn -> Application.put_env(:logflare, :supabase_mode, previous_supabase_mode) end) + on_exit(fn -> ConsolidatedSup.stop_pipeline(0) end) + + Logflare.Application.startup_tasks() + + user = SingleTenant.get_default_user() + + assert %Backend{ + id: 0, + token: nil, + user_id: user_id, + type: :clickhouse, + config: %{ + url: "http://localhost:8123", + database: "logflare_test", + port: 8123 + } + } = backend = Backends.get_default_backend(user) + + assert SingleTenant.default_clickhouse_backend?(backend) + refute SingleTenant.default_clickhouse_backend?(%{backend | id: 1}) + refute SingleTenant.default_clickhouse_backend?(%{backend | type: :postgres}) + assert user.id == user_id + assert SingleTenant.get_default_backend() == backend + assert Backends.get_backend(0) == backend + assert Backends.list_backends(user_id: user_id) == [] + end + end + describe "supabase_mode=true using Postgres" do TestUtils.setup_single_tenant( backend_type: :postgres, diff --git a/test/logflare/sources/source/supervisor_test.exs b/test/logflare/sources/source/supervisor_test.exs index 33669f8350..f7d99bf590 100644 --- a/test/logflare/sources/source/supervisor_test.exs +++ b/test/logflare/sources/source/supervisor_test.exs @@ -1,3 +1,25 @@ defmodule Logflare.Sources.Source.SupervisorTest do use Logflare.DataCase + + alias Logflare.Backends + alias Logflare.Backends.Adaptor.BigQueryAdaptor + alias Logflare.SingleTenant + alias Logflare.Sources.Cache + alias Logflare.Sources.Source.Supervisor + + test "runs the system backend source hook when source startup times out" do + source = build(:source) + + stub(SingleTenant, :single_tenant?, fn -> false end) + + expect(Cache, :get_by, fn [token: token] -> + assert token == source.token + source + end) + + expect(Backends, :start_source_sup, fn ^source -> {:error, :start_timeout} end) + expect(BigQueryAdaptor, :on_source_start, fn ^source -> :ok end) + + assert {:noreply, nil} = Supervisor.handle_cast({:create, source.token}, nil) + end end diff --git a/test/logflare/sources_test.exs b/test/logflare/sources_test.exs index 9dd03eba2e..1fa572bcb2 100644 --- a/test/logflare/sources_test.exs +++ b/test/logflare/sources_test.exs @@ -102,6 +102,33 @@ defmodule Logflare.SourcesTest do end end + describe "BigQuery source side effects with a single-tenant ClickHouse default" do + TestUtils.setup_single_tenant(backend_type: :clickhouse) + + setup do + insert(:plan, name: "Free", limit_source_ttl: :timer.hours(24) * 20) + [user: insert(:user)] + end + + test "creates a source without creating a BigQuery schema", %{user: user} do + assert {:ok, source} = Sources.create_source(%{name: TestUtils.random_string()}, user) + refute SourceSchemas.get_source_schema_by(source_id: source.id) + end + + test "updates TTL and clustering without patching a BigQuery table", %{user: user} do + reject(BigQuery, :patch_table_ttl, 4) + reject(BigQuery, :patch_table_clustering, 4) + + source = insert(:source, user: user, bigquery_clustering_fields: nil) + + assert {:ok, %Source{retention_days: 12, bigquery_clustering_fields: "session_id"}} = + Sources.update_source(source, %{ + retention_days: 12, + bigquery_clustering_fields: "session_id" + }) + end + end + describe "update_source/2 with different retention_days" do setup do user = insert(:user) @@ -474,6 +501,42 @@ defmodule Logflare.SourcesTest do end end + describe "Source.Supervisor with a single-tenant ClickHouse default" do + TestUtils.setup_single_tenant(backend_type: :clickhouse) + + setup do + reject(BigQuery, :init_table!, 6) + reject(BigQuery, :delete_table, 1) + + insert(:plan) + user = insert(:user) + + on_exit(fn -> + for {_id, child, _, _} <- DynamicSupervisor.which_children(Logflare.Backends.SourcesSup) do + DynamicSupervisor.terminate_child(Logflare.Backends.SourcesSup, child) + end + end) + + [user: user] + end + + test "starts and deletes a source without BigQuery table lifecycle calls", %{user: user} do + %{token: token} = source = insert(:source, user: user) + Sources.Cache.get_by_id(source.id) + start_supervised!(Source.Supervisor) + + assert {:ok, ^token} = Source.Supervisor.start_source(token) + TestUtils.retry_assert(fn -> assert {:ok, _pid} = Backends.lookup(SourceSup, source) end) + + assert {:ok, ^token} = Source.Supervisor.reset_source(token) + assert {:ok, ^token} = Source.Supervisor.delete_source(token) + + TestUtils.retry_assert(fn -> + assert {:error, :not_started} = Backends.lookup(SourceSup, source) + end) + end + end + test "ingest_ets_tables_started?/0" do assert Sources.ingest_ets_tables_started?() end @@ -550,6 +613,39 @@ defmodule Logflare.SourcesTest do end end + describe "shutdown_idle_sources/0 with a single-tenant ClickHouse default" do + TestUtils.setup_single_tenant(backend_type: :clickhouse) + + setup do + insert(:plan) + user = insert(:user) + source = insert(:source, user: user) + + backend = Backends.get_default_backend(user) + + Sources.Cache.get_by_id(source.id) + :ok = Backends.start_source_sup(source) + + {:ok, source: source, backend: backend} + end + + test "does not shut down a source while the consolidated default has pending events", %{ + source: source, + backend: backend + } do + event = build(:log_event, source: source) + queue_key = {:consolidated, backend.id} + Backends.IngestEventQueue.upsert_tid({:consolidated, backend.id, self()}) + Backends.IngestEventQueue.add_to_table(queue_key, [event]) + + assert Backends.system_default_ingest_queue_key(source) == queue_key + assert Backends.IngestEventQueue.total_pending(queue_key) == 1 + TestUtils.retry_assert(fn -> assert Backends.source_sup_started?(source) end) + :ok = Sources.shutdown_idle_sources() + assert Backends.source_sup_started?(source) + end + end + describe "list_sources/1" do setup do insert(:plan) diff --git a/test/logflare/sql_test.exs b/test/logflare/sql_test.exs index 9733ddfc28..8fb0eca78b 100644 --- a/test/logflare/sql_test.exs +++ b/test/logflare/sql_test.exs @@ -662,8 +662,7 @@ defmodule Logflare.SqlTest do # setup local clickhouse for e2e test {source, backend} = setup_clickhouse_test(source: source, user: user) - {:ok, _pid} = ClickHouseAdaptor.start_link(backend) - assert :ok = ClickHouseAdaptor.provision_ingest_tables(backend) + provision_clickhouse_tables!(backend) log_events = [ build_mapped_log_event( diff --git a/test/logflare_web/controllers/api/query_controller_test.exs b/test/logflare_web/controllers/api/query_controller_test.exs index c3d8339e8a..a34bf795a9 100644 --- a/test/logflare_web/controllers/api/query_controller_test.exs +++ b/test/logflare_web/controllers/api/query_controller_test.exs @@ -1,7 +1,6 @@ defmodule LogflareWeb.Api.QueryControllerTest do use LogflareWeb.ConnCase - alias Logflare.Backends.Adaptor.ClickHouseAdaptor alias Logflare.Backends.Adaptor.PostgresAdaptor alias Logflare.DataCase @@ -215,7 +214,6 @@ defmodule LogflareWeb.Api.QueryControllerTest do describe "backend_id parameter" do test "?sql= with backend_id uses backend's language", %{conn: conn, user: user} do {_source, backend} = DataCase.setup_clickhouse_test(user: user) - start_supervised!({ClickHouseAdaptor, backend}) query = "SELECT dummy AS my_time FROM system.one LIMIT 1 BY dummy" @@ -236,7 +234,6 @@ defmodule LogflareWeb.Api.QueryControllerTest do test "?ch_sql= with backend_id", %{conn: conn, user: user} do {_source, backend} = DataCase.setup_clickhouse_test(user: user) - start_supervised!({ClickHouseAdaptor, backend}) query = "SELECT dummy AS my_time FROM system.one LIMIT 1 BY dummy" @@ -253,8 +250,7 @@ defmodule LogflareWeb.Api.QueryControllerTest do conn: conn, user: user } do - {_source, backend} = DataCase.setup_clickhouse_test(user: user) - start_supervised!({ClickHouseAdaptor, backend}) + {_source, _backend} = DataCase.setup_clickhouse_test(user: user) query = "SELECT dummy AS my_time FROM system.one LIMIT 1 BY dummy" @@ -269,7 +265,6 @@ defmodule LogflareWeb.Api.QueryControllerTest do test "bq_sql param with backend_id executes query", %{conn: conn, user: user} do {_source, clickhouse_backend} = DataCase.setup_clickhouse_test(user: user) - start_supervised!({ClickHouseAdaptor, clickhouse_backend}) query = ~S|select 1 as my_time| @@ -284,7 +279,6 @@ defmodule LogflareWeb.Api.QueryControllerTest do test "?sql= takes precedence over deprecated params", %{conn: conn, user: user} do {_source, backend} = DataCase.setup_clickhouse_test(user: user) - start_supervised!({ClickHouseAdaptor, backend}) response = conn diff --git a/test/logflare_web/controllers/plugs/buffer_limiter_test.exs b/test/logflare_web/controllers/plugs/buffer_limiter_test.exs index efa2be64ff..62df1bf895 100644 --- a/test/logflare_web/controllers/plugs/buffer_limiter_test.exs +++ b/test/logflare_web/controllers/plugs/buffer_limiter_test.exs @@ -187,6 +187,40 @@ defmodule LogflareWeb.Plugs.BufferLimiterTest do %{source_id: ^source_id, source_token: ^source_token}} end + describe "single-tenant ClickHouse system default" do + TestUtils.setup_single_tenant(backend_type: :clickhouse) + + test "uses the consolidated default cache", %{ + conn: conn, + source: source + } do + backend = Backends.get_default_backend(%Logflare.User{id: source.user_id}) + + over_limit = Backends.max_buffer_queue_len() + 1 + + Logflare.PubSubRates.Cache.cache_buffers(source.id, nil, %{ + Node.self() => %{len: over_limit, queues: [{{source.id, nil, self()}, over_limit}]} + }) + + conn = conn |> assign(:source, source) |> BufferLimiter.call(%{}) + refute conn.halted + + assert Backends.system_default_ingest_queue_key(source) == {:consolidated, backend.id} + + Logflare.PubSubRates.Cache.cache_buffers(:consolidated, backend.id, %{ + Node.self() => %{ + len: over_limit, + queues: [{{:consolidated, backend.id, self()}, over_limit}] + } + }) + + conn = conn |> recycle() |> assign(:source, source) |> BufferLimiter.call(%{}) + + assert conn.halted + assert json_response(conn, 429) + end + end + describe "default ingest feature" do setup do user = insert(:user) @@ -511,14 +545,14 @@ defmodule LogflareWeb.Plugs.BufferLimiterTest do end # Fill CH backend queue over the limit - clickhouse_queue_key = {source.id, clickhouse_backend.id, self()} + clickhouse_queue_key = {:consolidated, clickhouse_backend.id, self()} IngestEventQueue.upsert_tid(clickhouse_queue_key) fill_queue(clickhouse_queue_key, source) # Cache buffer stats for both backends Backends.cache_local_buffer_lens(source.id, nil) - Backends.cache_local_buffer_lens(source.id, clickhouse_backend.id) + Backends.cache_local_buffer_lens(:consolidated, clickhouse_backend.id) conn = conn diff --git a/test/logflare_web/live_views/endpoints_live_test.exs b/test/logflare_web/live_views/endpoints_live_test.exs index 74d8ac9d92..963db84545 100644 --- a/test/logflare_web/live_views/endpoints_live_test.exs +++ b/test/logflare_web/live_views/endpoints_live_test.exs @@ -4,7 +4,7 @@ defmodule LogflareWeb.EndpointsLiveTest do use LogflareWeb.ConnCase import Logflare.ClickHouseMappedEvents, only: [build_mapped_log_event: 1] - import Logflare.DataCase, only: [setup_clickhouse_test: 1] + import Logflare.DataCase, only: [provision_clickhouse_tables!: 1, setup_clickhouse_test: 1] setup %{conn: conn} do insert(:plan) @@ -1160,8 +1160,7 @@ defmodule LogflareWeb.EndpointsLiveTest do {edge_source, backend} = setup_clickhouse_test(source: edge_source, user: user) - start_supervised!({ClickHouseAdaptor, backend}) - assert :ok = ClickHouseAdaptor.provision_ingest_tables(backend) + provision_clickhouse_tables!(backend) log_events = [ build_mapped_log_event( diff --git a/test/logflare_web/live_views/query_live_test.exs b/test/logflare_web/live_views/query_live_test.exs index 90826d89f6..8dbb9df72f 100644 --- a/test/logflare_web/live_views/query_live_test.exs +++ b/test/logflare_web/live_views/query_live_test.exs @@ -191,8 +191,6 @@ defmodule LogflareWeb.QueryLiveTest do {source, backend} = Logflare.DataCase.setup_clickhouse_test(user: user, source: source) - start_supervised!({ClickHouseAdaptor, backend}) - query = ~s(select non_existent from "#{source.name}") {:ok, view, _html} = diff --git a/test/support/data_case.ex b/test/support/data_case.ex index 4d86aa2e84..19e8008a71 100644 --- a/test/support/data_case.ex +++ b/test/support/data_case.ex @@ -16,7 +16,10 @@ defmodule Logflare.DataCase do alias Ecto.Adapters.SQL alias Logflare.Backends.Adaptor.ClickHouseAdaptor + alias Logflare.Backends.Adaptor.ClickHouseAdaptor.EndpointUtils + alias Logflare.Backends.Backend alias Logflare.Backends.ConsolidatedSup + alias Logflare.TestUtils using do quote do @@ -62,6 +65,8 @@ defmodule Logflare.DataCase do # judged against genuinely-still-live state — including other concurrently # running tests' own queues_keys — not a mapper this same callback is about # to clear out from under them. + IngestEventQueue.delete_stale_mappings() + for queues_key <- IngestEventQueue.list_generation_queues_keys(), IngestEventQueue.list_queues(queues_key) == [] do IngestEventQueue.prune_generations(queues_key) @@ -115,59 +120,38 @@ defmodule Logflare.DataCase do defdelegate errors_on(changeset), to: LogflareWeb.Utils, as: :changeset_errors @doc """ - Sets up a ClickHouse test environment with automatic cleanup. + Creates ClickHouse source and backend fixtures. - Returns `{source, backend}` tuple. Registers cleanup via `on_exit/1`. + Use `provision_clickhouse_tables!/1` to create tables with automatic teardown, + or `drop_clickhouse_tables_on_exit/1` when testing provisioning or ingestion startup. ## Options - `:config` - Custom ClickHouse configuration (merged with defaults) - `:user` - Existing user to use (creates one if not provided) - `:source` - Existing source to use (creates one if not provided) - - `:default_ingest_backend?` - Whether to set the backend as the default ingest backend (requires a source to be provided with the default ingest backend option set to true) - - `:cleanup?` - Whether to drop the backend's ClickHouse tables on exit (defaults to true) + - `:default_ingest?` - Whether to set the backend as the default ingest backend """ + @spec setup_clickhouse_test(keyword()) :: + {Logflare.Sources.Source.t(), Logflare.Backends.Backend.t()} def setup_clickhouse_test(opts \\ []) do config = Keyword.get(opts, :config, %{}) - default_ingest_backend? = Keyword.get(opts, :default_ingest_backend?, false) if not (is_map_key(config, :url) or is_map_key(config, :port)) do ensure_clickhouse_reachable!() end - user = - case Keyword.get(opts, :user) do - nil -> - Logflare.Factory.insert(:user) - - existing_user -> - existing_user - end - + user = Keyword.get(opts, :user) || Logflare.Factory.insert(:user) source = Keyword.get(opts, :source) || Logflare.Factory.insert(:source, user: user) - default_config = %{ - url: "http://localhost:8123", - database: "logflare_test", - username: "logflare", - password: "logflare", - port: 8123, - ingest_pool_size: 5, - query_pool_size: 3 - } - backend = Logflare.Factory.insert(:backend, type: :clickhouse, - config: Map.merge(default_config, config), - default_ingest?: default_ingest_backend?, + config: Map.merge(TestUtils.clickhouse_config(), config), + default_ingest?: Keyword.get(opts, :default_ingest?, false), user: user, sources: [source] ) - if Keyword.get(opts, :cleanup?, true) do - on_exit(fn -> cleanup_clickhouse_tables(backend) end) - end - {source, backend} end @@ -183,64 +167,56 @@ defmodule Logflare.DataCase do end end - @doc """ - Builds ClickHouse connection options for testing. - """ - def build_clickhouse_connection_opts(source, backend, type) when type in [:ingest, :query] do - base_opts = [ - scheme: "http", - hostname: "localhost", - port: 8123, - database: "logflare_test", - username: "logflare", - password: "logflare" - ] - - type_specific_opts = - case type do - :ingest -> [pool_size: 5, timeout: 15_000] - :query -> [pool_size: 3, timeout: 60_000] - end - - connection_name = - case type do - :ingest -> ClickHouseAdaptor.connection_pool_via({source, backend}) - :query -> ClickHouseAdaptor.connection_pool_via(backend) - end - - base_opts - |> Keyword.merge(type_specific_opts) - |> Keyword.put(:name, connection_name) + @spec provision_clickhouse_tables!(Backend.t()) :: :ok + def provision_clickhouse_tables!(backend) do + drop_clickhouse_tables_on_exit(backend) + assert :ok = ClickHouseAdaptor.provision_ingest_tables(backend) end @doc """ - Cleanup ClickHouse tables for a given `Backend`. - - Drops all type-specific tables (`_logs`, `_metrics`, `_traces`). + Registers teardown to synchronously drop a backend's typed tables after a test, + stopping its pipeline first. Uses the primary connection, not read routing. """ - def cleanup_clickhouse_tables(backend) do + @spec drop_clickhouse_tables_on_exit(Backend.t()) :: :ok + def drop_clickhouse_tables_on_exit(%Backend{config: config} = backend) do tables = - Enum.map([:log, :metric, :trace], fn type -> - ClickHouseAdaptor.clickhouse_ingest_table_name(backend, type) - end) - - drop_query = - tables - |> Enum.map_join("; ", fn table_name -> "DROP TABLE IF EXISTS #{table_name}" end) - - try do - ClickHouseAdaptor.execute_ch_query( - backend, - drop_query, - [], - pool_timeout: 1_000 + Enum.map( + [:log, :metric, :trace], + &ClickHouseAdaptor.clickhouse_ingest_table_name(backend, &1) ) + {scheme, hostname, port} = EndpointUtils.origin(config.url, config[:port]) + + opts = [ + scheme: scheme, + hostname: hostname, + port: port, + database: config.database, + username: config[:username], + password: config[:password], + pool_size: 1, + timeout: 60_000 + ] + + on_exit(fn -> ConsolidatedSup.stop_pipeline(backend.id) - rescue - _ -> :ok - catch - :exit, _ -> :ok + drop_clickhouse_tables(opts, tables) + end) + end + + @spec drop_clickhouse_tables(keyword(), [String.t()]) :: :ok + defp drop_clickhouse_tables(opts, tables) do + {:ok, conn} = Ch.start_link(opts) + + try do + Enum.each(tables, fn table -> + Ch.query!(conn, "DROP TABLE IF EXISTS #{table} SYNC", [], + timeout: 60_000, + pool_timeout: 60_000 + ) + end) + after + GenServer.stop(conn) end end diff --git a/test/support/test_utils.ex b/test/support/test_utils.ex index c7d7a7ccd8..0ccf899f3d 100644 --- a/test/support/test_utils.ex +++ b/test/support/test_utils.ex @@ -26,6 +26,7 @@ defmodule Logflare.TestUtils do Options: - :seed_user - boolean - enable to seed default plan and user - :supabase_mode - enable to seed supabase data + - :backend_type - the selected single-tenant backend """ defmacro setup_single_tenant(opts \\ []) do opts = @@ -42,8 +43,18 @@ defmodule Logflare.TestUtils do setup do initial_single_tenant = Application.get_env(:logflare, :single_tenant) + + initial_single_tenant_backend = + Application.get_env(:logflare, :single_tenant_backend) + Application.put_env(:logflare, :single_tenant, true) + Application.put_env( + :logflare, + :single_tenant_backend, + unquote(opts.backend_type) + ) + if unquote(opts.seed_user) do {:ok, _} = SingleTenant.create_default_plan() {:ok, _user} = SingleTenant.create_default_user() @@ -67,6 +78,13 @@ defmodule Logflare.TestUtils do on_exit(fn -> Application.put_env(:logflare, :single_tenant, initial_single_tenant) + + Application.put_env( + :logflare, + :single_tenant_backend, + initial_single_tenant_backend + ) + Application.put_env(:logflare, :public_access_token, initial_public_access_token) Application.put_env(:logflare, :private_access_token, initial_private_access_token) end) @@ -115,6 +133,34 @@ defmodule Logflare.TestUtils do end end + defp setup_single_tenant_backend(%{backend_type: :clickhouse}) do + quote do + setup do + previous = Application.get_env(:logflare, :clickhouse_backend_adapter) + + Application.put_env( + :logflare, + :clickhouse_backend_adapter, + Map.to_list(Logflare.TestUtils.clickhouse_config()) + ) + + on_exit(fn -> Application.put_env(:logflare, :clickhouse_backend_adapter, previous) end) + :ok + end + end + end + + @spec clickhouse_config() :: map() + def clickhouse_config do + %{ + url: "http://localhost:8123", + database: "logflare_test", + username: "logflare", + password: "logflare", + port: 8123 + } + end + def default_bq_schema, do: SchemaBuilder.initial_table_schema() def build_bq_schema(params) do