feat(outputs.zerobus): Add plugin - #19444
Conversation
telegraf-tiger
Bot
added
feat
srebhan
left a comment
srebhan
left a comment
There was a problem hiding this comment.
Thanks for the minimal starting point @zlata-stefanovic-db! Please find my comments below. Altogether, I think there is no deal-breaker issue here so looking forward to your update!
Address the first review notes: link Unity Catalog Delta, shorten the intro, rename endpoint/workspace/table/application/timeout, let application override the product token, and nest mapping/batching under Configuration.
Document tag/field name conflicts under metric mapping, drop the global omit_hostname tip, and note that NaN rejection comes from the JSON ingest path rather than Delta itself.
Unroll required-option checks, stop trimming column names, only compare timestamp/measurement when both are set, and inline the default timeout. Clarify SDK vs Zerobus limits in the README.
Keep the default timeout only in the constructor, treat timeout=0 as no timeout, defer context cancels, and drop redundant interface comments.
Use one context for the schema fetch and the stream creation so the timeout covers all of startup, drop the stream options that only repeat SDK defaults, and set the stream and columns only once both succeed.
Return the startup error alone and only log a failing close, and drop the SDK nil-check in Write as the agent never writes before a successful connect.
…them Closing a stream that is already broken is expected to fail and no caller can act on it, so report it in the log and keep the returned errors about what actually went wrong.
Handle the serialization failure and the oversized record each in their own block so the size check no longer hides behind a success check and the accepted path stays at the top level of the loop.
Both byte constants described the budget for one ingest request, so fold them into a single maxRequestBytes instead of implying a per-record cap.
…trics Split the metrics into requests as they are serialized and keep the metric indices with each request, so a failing request can later be mapped back to the metrics it carries.
The test needs a live Databricks workspace, so it can never run in CI and carries no value for the project. A real integration test needs a fake ingest server, which cannot be built outside the SDK today.
The test only asserted that Go assigns the values given at registration.
Declare the configuration per test case instead of mutating a shared helper, and cover the timeout and column settings in separate tables so each case shows the full input it validates.
Compare the serialized record against one expected JSON document, and use metrics the agent can actually produce instead of overriding FieldList to inject values that metric construction rejects.
Telegraf itself cannot produce nil or unsupported fields, so record why the serializer still rejects them instead of assuming the input shape.
|
Hi @srebhan thank you for the detailed review! I implemented everything that's possible right now, here are some things to follow-up on/get your opinion about. Writing it here for easier tracking: Deferred to follow-up PRs
Waiting on your call
Kept as is
|
|
Let's go from your current state. Regarding the type-guard: Keep it if you insist, the metric interface is only used with the metric implementation(s) in Telegraf and those are safe but if you feel better with the guard... ;-) Will review your code shortly... |
Returning the partial write error suggested the plugin handles partial writes, which it does not yet. The rejects are logged instead and the retry behavior of a failing write is stated in the code.
Serializing a metric says nothing about whether the endpoint stored it, so claiming it as accepted would drop metrics from the agent buffer that a failing write never delivered.
The cases shared one record slice and derived the size limit from it, so a case could affect the others and the limit was hidden behind a variable.
srebhan
left a comment
srebhan
left a comment
There was a problem hiding this comment.
Very nice @zlata-stefanovic-db! Three small comments on error messages left... Can you please have a look!?
Also state in the log that the rejected metrics are dropped.
srebhan
left a comment
srebhan
left a comment
There was a problem hiding this comment.
Awesome! Thanks for making this PR a joy to review @zlata-stefanovic-db!
srebhan
added
the
ready for final review
Closing is what makes the SDK forget a stream, so replacing a terminally failed one without closing kept it and its unacknowledged records around until the agent shut down.
A mistyped column name was silently ignored, so every row was written without the timestamp or the measurement name.
The "timeout" option only covers opening the stream, so state the SDK timeouts a user is subject to while writing.
Warning about it kept writing metrics without their timestamp, so error out and point at "timestamp_column" for tables not storing a timestamp.
Both error paths aborting the stream startup were untested, so check the extracted columns as well as an unparseable and an empty descriptor.
skartikey
left a comment
skartikey
left a comment
There was a problem hiding this comment.
@zlata-stefanovic-db Thanks for the quick turnaround on this. I have a couple more comments before we can close it out. Please take a look.
| // together with the valid metrics. | ||
| records, err := z.serializeMetrics(metrics) | ||
| if err != nil { | ||
| z.Log.Errorf("Serializing metrics failed: %v; dropping the rejected metrics", err) |
There was a problem hiding this comment.
serializeMetrics builds a full PartialWriteError and we drop it here, so Write returns nil at line 148 even when we discarded metrics. I ran this against RunningOutput: with nil, every tracking metric in the batch reports Delivered() == true, including the dropped one, so a tracking input acks a metric that never reached Databricks. Return the error instead, with MetricsAccept set to the indices not in MetricsReject, which after a successful Flush is exactly what was acknowledged. Fill both lists though: with reject only, the written metrics land in neither and InferKeep puts them back in the buffer.
There was a problem hiding this comment.
Fixed: serializeMetrics now also returns the indices it serialized, and Write returns the PartialWriteError after a successful flush with MetricsAccept set to those and MetricsReject as before, so nothing falls into InferKeep. The early return nil when every metric was rejected had the same problem and is gone too, and I dropped the Log.Errorf since the agent already logs what Write returns.
Failure paths still return a plain error, so the whole batch is kept and the rejects are dropped on the next successful write. Happy to reject them right away if you prefer.
| // rather than assumed away. | ||
| func metricToTableSchemaJSON(metric telegraf.Metric, timestampColumn, measurementColumn string, columns map[string]bool) ([]byte, error) { | ||
| values := make(map[string]interface{}, len(metric.TagList())+len(metric.FieldList())+2) | ||
| if timestampColumn != "" && columns[timestampColumn] { |
There was a problem hiding this comment.
[nit] openStream now rejects a column the table doesn't have, so this lookup and the one on line 26 are dead outside the unit tests. Dropping them makes it clear openStream owns that check.
There was a problem hiding this comment.
You're right, dropping the lookups.
| }, | ||
| } | ||
|
|
||
| plugin := &Zerobus{ |
There was a problem hiding this comment.
[nit] Still shared by every subtest, same as last round. Moving it inside the t.Run makes each case self-contained.
| // The byte budget applies to all records of a request together and reserves | ||
| // headroom for the surrounding request fields. | ||
| z.maxRecords = 100000 | ||
| z.maxBytes = 10*1024*1024 - 64*1024 - 1024 |
There was a problem hiding this comment.
[question] Where does the 65 KiB of headroom come from? The 10 MiB and the 100k records are named in the comment above, but not why it's 64 KiB plus 1 KiB. A link or a note that it's a measured worst case would stop the next person tuning it blind.
There was a problem hiding this comment.
The 64 KiB is the SDK's own margin below the service limit — stream.DefaultMaxPayloadBytes = 10 * 1024 * 1024 - 64 * 1024, commented there as "leaves room below the 10 MiB service limit". It lives in the SDK's internal/ package, so the plugin can't import it and has to repeat the value.
The kilobyte on top is mine: recordSize only counts the tag and length prefix of each record, while the request itself carries fields around them — the SDK estimates that at 512 bytes per request plus 32 per record. Not a measured worst case, just a round number above the SDK's own estimate. Noted both in the comment.
Returning nil made Telegraf accept the whole batch, so tracking metrics were acknowledged although the plugin dropped them. Report the rejected metrics and accept the ones the endpoint acknowledged instead.
Opening the stream fails on a column the table does not have, so looking the same columns up again while serializing is dead code.
A shared instance made the cases depend on each other's setup.
Neither number was arbitrary, but nothing said where they came from.
|
Download PR build artifacts for linux_amd64.tar.gz, darwin_arm64.tar.gz, and windows_amd64.zip. 📦 Click here to get additional PR build artifactsArtifact URLs |
skartikey
left a comment
skartikey
left a comment
There was a problem hiding this comment.
@zlata-stefanovic-db One last comment(linter failure)
revive enforce-slice-style treats []int{} as nil, while serializeMetrics
returns a non-nil empty slice.
skartikey
left a comment
skartikey
left a comment
There was a problem hiding this comment.
@zlata-stefanovic-db Thanks for the contribution!
3 participants
Summary
Add an output plugin that writes metrics to a Unity Catalog Delta table using the
Databricks Zerobus Ingest service, which accepts records over gRPC and
commits them into Delta directly.
Metrics can already reach Zerobus through the generic
httpoutput by posting JSONrecords to its REST endpoint. A dedicated plugin uses the streaming gRPC API through
a Databricks Go SDK instead, so OAuth is handled for you, the destination table
layout is derived from Unity Catalog rather than assembled by hand in the
configuration, and large batches are considerably faster.
Tags and fields are written to same-named columns of the destination table. The
metric timestamp goes to
timestamp_columnand the measurement name tomeasurement_column, both optional. The table schema is read from Unity Catalogwhen a stream is opened, so columns added by an
ALTER TABLEare picked up withoutrestarting Telegraf.
Batches are split into requests that stay inside the Zerobus size limits and the
write succeeds only once Databricks has acknowledged every record. A metric that
cannot be encoded, for example because a tag and a field share a name, is rejected
on its own so it cannot stall the buffer. A failing write returns an error and
Telegraf retries the buffered batch on a new stream, which makes the destination
table at-least-once.
This is the bare minimum plugin split out of #19441 following the review there. The
static schema mode, the concurrent streams and the resume-after-partial-ack retry
behaviour are left out, and the client is constructed directly instead of behind an
interface for faking. I would like to propose those features in follow-up pull
requests.
Dependencies
Adds
github.com/databricks/zerobus-sdk/purego v0.1.0(Apache 2.0), a DatabricksZerobus Go SDK.
docs/LICENSE_OF_DEPENDENCIES.mdis updated accordingly.Tests
Unit tests cover the default and required options, option validation, the record
encoding and the values it rejects, the partial-write error for metrics that cannot
be written, and the splitting of a batch into requests. An integration test writes
to a real Databricks workspace when credentials are provided.
Checklist
I have signed the InfluxData CLA.
Related issues
No prior issue exists for this plugin; this pull request introduces it directly.