feat: logstash ingestion - #3940
FranekKubis wants to merge 7 commits into
Conversation
FranekKubis
changed the title
b354054 to
fe7b7d4
Compare
| %{ | ||
| url: config.url, | ||
| http: "http1", | ||
| gzip: Map.get(config, :gzip, true), |
There was a problem hiding this comment.
we should always set this to true and remove it as a configurable.
There was a problem hiding this comment.
We can remove it from the otlp drain as well, don't see a point in letting users configure it
| password: :string, | ||
| # logstash, otlp | ||
| gzip: :boolean, | ||
| headers: {:map, :string}, |
There was a problem hiding this comment.
We can extend filebeat to handle the headers as well do that this is a common config across all 3
| headers: {:map, :string}, | ||
| # otlp | ||
| endpoint: :string, | ||
| protocol: :string |
There was a problem hiding this comment.
Protocol can be removed as a config since only one is supported.
| gzip: :boolean, | ||
| headers: {:map, :string}, | ||
| # otlp | ||
| endpoint: :string, |
There was a problem hiding this comment.
Hmm wonder if it's worth converting this to use :url so that it's consistent across all 3...thoughts?
There was a problem hiding this comment.
I would rather keep it as is because endpoint it is OTLP's own vocabulary (e.g. OTEL_EXPORTER_OTLP_ENDPOINT / OTEL_EXPORTER_OTLP_LOGS_ENDPOINT) and it is not quite a synonym for url.
|
|
||
| @impl Logflare.Backends.Adaptor | ||
| def test_connection(_), do: {:error, :not_implemented} | ||
| defp redact_header(k, v) do |
There was a problem hiding this comment.
There some common header redaction logic across the backends, should move them to a utility function under the HttpBased module
| "retry": {:hex, :retry, "0.19.0", "aeb326d87f62295d950f41e1255fe6f43280a1b390d36e280b7c9b00601ccbc2", [:mix], [], "hexpm", "85ef376aa60007e7bff565c366310966ec1bd38078765a0e7f20ec8a220d02ca"}, | ||
| "rustler": {:hex, :rustler, "0.37.3", "5f4e6634d43b26f0a69834dd1d3ed4e1710b022a053bf4a670220c9540c92602", [:mix], [{:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "a6872c6f53dcf00486d1e7f9e046e20e01bf1654bdacc4193016c2e8002b32a2"}, | ||
| "rustler_precompiled": {:hex, :rustler_precompiled, "0.8.2", "5f25cbe220a8fac3e7ad62e6f950fcdca5a5a5f8501835d2823e8c74bf4268d5", [:mix], [{:castore, "~> 0.1 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: false]}, {:rustler, "~> 0.23", [hex: :rustler, repo: "hexpm", optional: true]}], "hexpm", "63d1bd5f8e23096d1ff851839923162096364bac8656a4a3c00d1fff8e83ee0a"}, | ||
| "rustler_precompiled": {:hex, :rustler_precompiled, "0.9.0", "3a052eda09f3d2436364645cc1f13279cf95db310eb0c17b0d8f25484b233aa0", [:mix], [{:rustler, "~> 0.23", [hex: :rustler, repo: "hexpm", optional: true]}], "hexpm", "471d97315bd3bf7b64623418b3693eedd8e47de3d1cb79a0ac8f9da7d770d94c"}, |
There was a problem hiding this comment.
No deps changes, should remove this from diff
| ) | ||
|
|
||
| start_supervised!({AdaptorSupervisor, {source, backend}}) | ||
| :timer.sleep(500) |
There was a problem hiding this comment.
Should wrap an assertion with TestUtils retry_assert if there's a race condition and we need to wait for proc to startup
|
|
||
| assert html =~ "Logstash URL" | ||
| refute html =~ "Filebeat URL" | ||
| refute html =~ "Elastic Supabase Endpoint" |
There was a problem hiding this comment.
The otlp endpoint is bundled with elasticsearch actually, not supabase specific.
| refute html =~ "Elastic Supabase Endpoint" | |
| refute html =~ "Elasticsearch OTLP Endpoint" |
|
Could you also add a demo video of the setup + ingestion? 🙏 |
|
I am adding a rough recording of clicking my way through the app, initializing new backend for logstash and sending logs to the logflare source connected to the logstash backend. |
00d9e43 to
48e4307
Compare
![depthfirst-app[bot]](https://avatars.githubusercontent.com/in/1021002?s=60&v=4){kind=small}
62f6107 to
b800ab0
Compare
3 participants
Closes: O11Y-2422
Add logstash transport to elastic backend
Adds a third
transportvalue to the:elasticbackend, delivering events to aLogstash
httpinput as ECS-flavoured JSON.timestampandevent_messagearelifted to
@timestampandmessageso a receiving pipeline needs nodatefilter; remaining fields stay top-level and Logflare metadata is namespaced
under
logflare. Reuses WebhookAdaptor (HTTP/1.1, optional basic auth, gzip)via the existing
format_batchhook.Also fixes three pre-existing issues surfaced while verifying end to end:
ElasticAdaptor.validate_config/1had no catch-all clause, so an unsupportedtransport raised
CaseClauseErrorinstead of returning an invalid changeset.The existing "rejects unknown transport" test was already failing on this.
Logflare.FinchDefaultHttp1was never started in single-tenant Postgres mode,so every WebhookAdaptor-based backend (elastic, loki, webhook) crashed with
"unknown registry" on delivery.
The source edit page hardcoded which backend types it would render, hiding
elastic, loki, clickhouse, s3, axiom, otlp, incidentio and last9 and making
them impossible to attach to a source. Backend types and labels now derive
from a single registry in
Logflare.Backends.Backend, with a compile-timecheck that every adaptor has a label. Side effect: the "New backend" dropdown
is now ordered alphabetically.
Adds a
logstashdocker-compose service andtest/logstash.conffor localverification. Note that outbound requests are SSRF-protected, so a local
Logstash must be reached through a tunnel rather than
localhost.