Consolidated Ingest - Phase 5 (Supervision Tree/Worker) #3088
Conversation
…o-support-consolidated-ingest
…olidated backends in source supervisor
…lidated-ingestion
…lidated-ingestion
…gle implementation for now. This is temporary until PR#3089 is in place
| end | ||
|
|
||
| @impl GenStage | ||
| def format_discarded(discarded, %{consolidated: true} = state) do |
There was a problem hiding this comment.
we can consider removing this callback fully since we are not inserting directly into the Producer process anymore.
|
|
||
| case IngestEventQueue.pop_pending(key, n) do | ||
| {:error, :not_initialized} -> | ||
| Logger.warning( |
There was a problem hiding this comment.
| Logger.warning( | |
| Logger.error( |
| alias Logflare.Backends.Adaptor | ||
| alias Logflare.Backends.ConsolidatedSup | ||
|
|
||
| @default_interval 30_000 |
There was a problem hiding this comment.
we can consider moving the reconciliation logic to a stateless periodic job (hourly?) using quantum, since this 30s is high for the moduledoc goals (bullet point 2), as pt2 should happen very infrequently.
pt1 should already be handled on ingestion, a 30s wait would be too slow anyway.
| def start_pipeline(%Backend{} = backend) do | ||
| adaptor_module = Adaptor.get_adaptor(backend) | ||
| child_spec = adaptor_module.child_spec(backend) | ||
| DynamicSupervisor.start_child(@dynamic_sup_name, child_spec) |
There was a problem hiding this comment.
note: will probably need to partition this in future to prevent blocking, since DynamicSupervisor can be a bottleneck.
| ) | ||
| end | ||
|
|
||
| @spec build_telemetry_metadata(state :: DynamicPipeline.state()) :: %{ |
There was a problem hiding this comment.
fine for now, we can make this module more agnostic (and also easier to open source in the future) by hoisting the telemetry metadata building and/or callback definition to the caller.
| Logger.warning("Failed to start consolidated pipeline", | ||
| backend_id: backend.id, | ||
| reason: inspect(reason) | ||
| ) |
There was a problem hiding this comment.
| Logger.warning("Failed to start consolidated pipeline", | |
| backend_id: backend.id, | |
| reason: inspect(reason) | |
| ) | |
| Logger.warning("Failed to start consolidated pipeline: #{inspect(reason)}", | |
| backend_id: backend.id | |
| ) |
| if Adaptor.consolidated_ingest?(backend) do | ||
| case ConsolidatedSup.stop_pipeline(backend) do | ||
| :ok -> | ||
| Logger.info("Stopped consolidated pipeline", backend_id: backend.id) | ||
|
|
||
| {:error, :not_found} -> | ||
| :ok | ||
| end | ||
| end |
Split from #3084
See #3090 before merging this
BufferProducerNote that this is based off the branch in #3083 - so it looks a bit larger than it really is.