feat: emit advanced-tier forwarder.* custom metrics - #48
Conversation
Implements the advanced metrics_tier from the OCI Log Forwarder observability
design doc: 11 metrics on top of the 6 already shipped in basic (byte volumes,
decode/serialize errors, batching behavior, delivery error classes, run
duration, secret-fetch failures, client-cache hit rate). Opt-in via
FORWARDER_METRICS_TIER=advanced; none/basic customers are unaffected.
forwarder.delivery.retries is intentionally not implemented -- the design doc
itself conditions it on "if/when client retry is added", and no retry
mechanism exists in this codebase to observe. The suggested OCI-specific
dimensions (compartment name, log_group, log_source_type) are also deferred:
compartment name needs a new OCI Identity API dependency, which is meaningfully
separate scope.
bytes.delivered reuses the batch size already computed once while building the
batch (via a new BatchMessage wrapper carrying it through the channel) rather
than re-marshaling the whole batch again at delivery time.
Verified against a real OCI Function + New Relic account: all 11 metrics land
with correct values, including the error paths (decode.errors, and an
invocation that panics correctly recording invocations{status=error} with
nothing else counted).
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
…name/ application_name dimensions, which now live on the base branch (PR #47) instead of duplicated here. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
| // an oversized record arriving after another batch was already flushed would slip | ||
| // through uncounted. | ||
| if logSize > maxPayloadSize { | ||
| rec.Count(metrics.TierAdvanced, "forwarder.records.oversized", 1, nil) |
There was a problem hiding this comment.
Should we allow size more than 1MB?Will it not cause any issue at new relic side?
There was a problem hiding this comment.
This is intentional, best-effort behavior rather than a gap: MaxPayloadSize (1MB) is New Relic's documented Log API limit (https://docs.newrelic.com/docs/logs/log-api/introduction-log-api/#limits). An oversized single record is still attempted so we don't silently drop data, but it's flagged via forwarder.records.oversized here, and if NR's API does reject it, that already surfaces via forwarder.delivery.errors/forwarder.records.dropped in newrelic_client_util.go. So the observability is already in place — happy to revisit if we'd rather actively drop/truncate oversized records instead of attempting delivery.
There was a problem hiding this comment.
Keep close eye on observability.We dont want to silently drop also we dont want api to fail.
| }, | ||
| Entries: currentBatch, | ||
| }} | ||
| func ProduceMessageToChannel(channel chan BatchMessage, currentBatch common.LogData, attributes common.LogAttributes, sizeBytes int) { |
There was a problem hiding this comment.
channel <- BatchMessage{...} is an unconditional blocking send with no select/timeout/cancellation path. If the channel is full and nothing is draining it (e.g. all 6 workers stuck on a slow NR API call), this call just blocks until the invocation itself times out — there's no way for it to bail early or signal that it's stuck.
Lets try to handle it well.
There was a problem hiding this comment.
Fixed in 7e7d0e3: ProduceMessageToChannel now selects on ctx.Done() alongside the channel send and returns a bool, with ctx threaded through ProcessLogs/splitLogsIntoBatches/produceBatch from the invocation's own context. If the send can't complete because ctx is cancelled, we stop processing and record the affected records via forwarder.records.dropped (reason producer_cancelled) instead of blocking forever. Added a test (TestProduceMessageToChannel_ContextCancelled) covering the bail-out path.
ProduceMessageToChannel now selects on ctx.Done() alongside the channel send, and the loggroup callers thread ctx through so a full channel with no draining consumer no longer hangs until the OCI function's own hard timeout kills the invocation. Records that couldn't be handed off are now counted via forwarder.records.dropped instead of silently lost.
| nrClient, err := util.NewNRClient(rec) | ||
| if err != nil { | ||
| rec.Count(metrics.TierAdvanced, "forwarder.secret.fetch.errors", 1, nil) | ||
| log.Panicf("error initializing newrelic client: %v", err) |
There was a problem hiding this comment.
This line increments this metric on any non-nil error from NewNRClient, but that error can come from two unrelated places ,a bad NEW_RELIC_REGION (SetRegion failing) or an actual Vault fetch failure. A region misconfig gets mislabeled as a secret-fetch problem. Worse: once one real failure happens, the client-cache (lines 88-95) re-serves that same stale error on every invocation for up to 10 minutes (CLIENT_TTL), and each one re-increments the counter — one failure looks like a sustained outage.
| log.Panicf("error initializing newrelic client: %v", err) | ||
| } | ||
|
|
||
| runStart := time.Now() |
There was a problem hiding this comment.
This one starts the timer after util.NewNRClient(rec) returns. On a cold client-cache, that call does a real network round-trip to OCI Vault. So the metric named "run duration" excludes exactly the part most likely to be slow — and if NewNRClient errors, log.Panicf (line 57) fires before the deferred timer is even registered, so the slowest/failing invocations contribute zero data points.
Fix: move runStart := time.Now() above the NewNRClient call.
| rec.Summary(metrics.TierBasic, "forwarder.pipeline.lag", time.Since(env.Time).Seconds(), nil) | ||
| } | ||
|
|
||
| logBytes, err := json.Marshal(logData) |
There was a problem hiding this comment.
Records that fail to marshal vanish from the loss-accounting entirely.A record that fails json.Marshal is continued out and only counted in forwarder.serialize.errors, never in forwarder.records.dropped. Anyone reconciling received == delivered + dropped sees an unexplained gap and could wrongly conclude nothing was lost.
| rec.Count(metrics.TierBasic, "forwarder.records.dropped", float64(remaining), map[string]interface{}{"reason": "producer_cancelled"}) | ||
| log.Warnf("context cancelled while producing log batch; dropping %d remaining log record(s)", remaining) | ||
| return | ||
| } |
There was a problem hiding this comment.
produceBatch counts the batch it failed to send; the caller separately hand-computes
remaining := len(logs) - i
for the not-yet-batched tail. Correct today, but a future caller of produceBatch (a second batching path, a refactor) would naturally write if !produceBatch(...) { return } and get only half the accounting, silently undercounting drops — with zero test coverage on this exact path to catch it.
| if !util.ProduceMessageToChannel(ctx, channel, batch, commonAttributes, batchSize) { | ||
| rec.Count(metrics.TierBasic, "forwarder.records.dropped", float64(len(batch)), map[string]interface{}{"reason": "producer_cancelled"}) | ||
| log.Warnf("context cancelled while producing log batch; dropped %d log record(s)", len(batch)) | ||
| return false |
There was a problem hiding this comment.
| if len(currentBatch) == 0 { | ||
| currentBatch = common.LogData{logData} | ||
| currentBatchSize = logSize | ||
| } else if currentBatchSize+logSize > maxPayloadSize && len(currentBatch) > 0 { |
There was a problem hiding this comment.
that second clause is always true there (the prior if len(currentBatch) == 0 branch already ruled out the empty case). Dead condition, safe to drop.
3 participants
Summary
Stacked on top of #47 (basic-tier
forwarder.*metrics). Implements theadvancedtier from the OCI Log Forwarder observability design doc: 11 more metrics on top of the 6 already shipped inbasic, opt-in viametrics_tier = "advanced".none/basicbehavior and cost are completely unaffected — every new metric is gated the same way the existing 6 already are.What's added
forwarder.bytes.receivedUnmarshalforwarder.decode.errors(error_class)forwarder.batches.created/forwarder.batch.size_bytesforwarder.records.oversizedmaxPayloadSizeforwarder.serialize.errorsforwarder.bytes.deliveredforwarder.delivery.errors(error_class,status)records.droppedforwarder.run.durationforwarder.secret.fetch.errorsforwarder.client.cache(`result=hitTwo things intentionally left out, both explained in code comments:
forwarder.delivery.retries— the design doc itself conditions this on "if/when client retry is added"; there's no retry mechanism in this codebase to observe, so faking it would give false confidence.Testing
go build/go vet/go test ./... -raceall clean.Also deployed to a real OCI Function + New Relic account (not just unit tests) and confirmed all 11 metrics land with correct values — including the error paths: an invocation that panics on bad input correctly records
invocations{status=error}with nothing else counted, anddecode.errorsfires with the righterror_class.Review
Went through this as a self-review pass before opening the PR: found and fixed one real bug (
records.oversizedwas only checked when a record happened to start a fresh batch, so an oversized record arriving after another batch was already flushed slipped through uncounted — fixed to check unconditionally per record, with a regression test), plus a redundant nil guard, a duplicated comment, and thebytes.delivereddouble-marshal mentioned above.