Skip to content

BQ storage write api - Mint pool #3272

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Ziinc merged 11 commits into from
Apr 15, 2026
Merged

BQ storage write api - Mint pool #3272

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
11ef3f2
feat: Add gRPC supervised channel pool for Mint adapter
bblaszkow06 Mar 6, 2026
a073e67
Add tests for pool, interceptor and channel monitor
bblaszkow06 Mar 11, 2026
d5ea6cf
Improve channel monitor tests
bblaszkow06 Mar 12, 2026
2142b78
fix: GRPC.Server.Supervisor setup
bblaszkow06 Mar 19, 2026
29a792d
fix: Leaking channel processes
bblaszkow06 Mar 19, 2026
6795557
fix: proper response stream handling
bblaszkow06 Mar 19, 2026
a2990f0
temp: Compare implementations
bblaszkow06 Apr 2, 2026
f053d7a
Remove Finch-based GRPC Write API, fix tests
bblaszkow06 Mar 19, 2026
b2ac6d7
feat: Make Grpc Write API pool size configurable
bblaszkow06 Mar 19, 2026
2eb7d70
fix: apply lint rule
bblaszkow06 Mar 25, 2026
1d15a33
fix tests
bblaszkow06 Mar 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion config/runtime.exs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,8 @@ config :logflare,
encryption_key_retired: System.get_env("LOGFLARE_DB_ENCRYPTION_KEY_RETIRED"),
metadata: logflare_metadata,
health: logflare_health,
http_connection_pools: http_connection_pools
http_connection_pools: http_connection_pools,
bq_write_api_pool_size: System.get_env("LOGFLARE_BQ_WRITE_API_POOL_SIZE")
Copy link
Copy Markdown
Contributor

@Ziinc Ziinc Apr 15, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
bq_write_api_pool_size: System.get_env("LOGFLARE_BQ_WRITE_API_POOL_SIZE")
bq_write_api_pool_size: System.get_env("LOGFLARE_BIGQUERY_WRITE_API_POOL_SIZE")

In my opinion we should automatically manage connection pool. But we can give this a try first. Needs a sensible default too.

]
|> filter_nil_kv_pairs.()

Expand Down
5 changes: 4 additions & 1 deletion lib/logflare/application.ex
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,7 +103,10 @@ defmodule Logflare.Application do
LogflareWeb.Endpoint,
# If we get a log event and the Source.Supervisor is not up it will 500
{GRPC.Server.Supervisor,
endpoint: LogflareGrpc.Endpoint, port: grpc_port, cred: grpc_creds, start_server: true},
endpoint: LogflareGrpc.Endpoint,
port: grpc_port,
start_server: true,
adapter_opts: [cred: grpc_creds]},
# Monitor system level metrics
SystemMetricsSup,
Logflare.Telemetry,
Expand Down
7 changes: 2 additions & 5 deletions lib/logflare/backends/adaptor/bigquery_adaptor.ex
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,6 @@ defmodule Logflare.Backends.Adaptor.BigQueryAdaptor do
end

def insert_log_events_via_storage_write_api(log_events, opts) do
# convert log events to table rows
context =
Keyword.validate!(opts, [
:project_id,
Expand All @@ -115,7 +114,6 @@ defmodule Logflare.Backends.Adaptor.BigQueryAdaptor do
:backend_id
])

# get table id
table_id = format_table_name(opts[:source_token])

arrow_data =
Expand Down Expand Up @@ -143,14 +141,13 @@ defmodule Logflare.Backends.Adaptor.BigQueryAdaptor do
OpenTelemetry.Tracer.with_span "ingest.bq_api_call", %{
attributes: %{insert_method: :bq_storage_write}
} do
GoogleApiClient.append_rows({:arrow, arrow_data}, context, table_id)
|> tap(fn
case GoogleApiClient.append_rows({:arrow, arrow_data}, context, table_id) do
{:error, reason} ->
OpenTelemetry.Tracer.set_status(:error, inspect(reason))

_ ->
:ok
end)
end
end
end

Expand Down
101 changes: 0 additions & 101 deletions lib/logflare/backends/adaptor/bigquery_adaptor/finch/bidirectional_stream.ex
View file
Open in desktop

This file was deleted.

28 changes: 0 additions & 28 deletions lib/logflare/backends/adaptor/bigquery_adaptor/finch/custom_stream.ex
View file
Open in desktop

This file was deleted.

Loading
Loading