feat: add structured logging and observability hooks by intel352 · Pull Request #7 · GoCodeAlone/workflow-plugin-bento

intel352 · 2026-02-23T10:18:29Z

Summary

Add structured logging via log/slog for all module lifecycle events (stream start/stop/error, message processed, topic events, processing start/complete/error)
Add per-stream metrics tracking (messages in, messages out, errors, uptime, last message time) using sync/atomic for thread safety
Add health status reporting per stream (healthy / degraded / unhealthy) based on running state and error count
Integrate logging, metrics, and health into stream_module, input_module, output_module, broker_module, and processor_step
Fix three pre-existing lint issues: unchecked MetaWalkMut return, unused mapToMessage helper, and nolint annotation on future-use ensureStream

Closes #3

Test plan

Unit tests for logger structured output format (7 tests)
Unit tests for metrics counting (MessagesIn, MessagesOut, Errors, Uptime, LastMessageTime, Snapshot)
Unit tests for health status transitions (healthy → degraded → unhealthy)
Tests for concurrent metric updates (50 goroutines × 100 messages, no races)
All 21 tests pass with -race flag
golangci-lint run reports 0 issues
go build passes

🤖 Generated with Claude Code

Copilot

Pull request overview

This PR adds comprehensive structured logging and observability infrastructure to the Bento workflow plugin. It introduces standardized logging via log/slog, thread-safe metrics tracking using atomic operations, and health status reporting based on runtime state and error counts. The implementation follows a consistent pattern across all module types (stream, input, output, broker, processor) and includes comprehensive unit tests.

Changes:

Added structured logging infrastructure using log/slog with consistent patterns for lifecycle events, message throughput, and errors
Implemented thread-safe metrics tracking (messages in/out, errors, uptime) using sync/atomic for counters and mutex-protected timestamps
Added health status reporting (healthy/degraded/unhealthy) with per-module Health() methods returning detailed reports
Integrated observability hooks into all module types and processor steps
Fixed three pre-existing lint issues (unchecked return value, unused function, missing nolint annotation)
Added 21 comprehensive unit tests covering logging output, metrics counting, health transitions, and concurrent access

Reviewed changes

Copilot reviewed 10 out of 10 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
internal/logger.go	New structured logger wrapper using slog with component-specific context fields and methods for stream lifecycle, message processing, and error logging
internal/metrics.go	New thread-safe metrics tracker using atomic counters for messages/errors and mutex for timestamps, with snapshot capability
internal/health.go	New health status tracker with three states (healthy/degraded/unhealthy) based on running state and error threshold
internal/stream_module.go	Integrated logging, metrics, and health tracking; added Health() method
internal/input_module.go	Integrated observability with message counting on publish; added Health() method
internal/output_module.go	Integrated observability with message counting on forward; added Health() method
internal/broker_module.go	Integrated observability across per-topic streams; added Health() method
internal/processor_step.go	Added logging for processing lifecycle (start/complete/error)
internal/bridge.go	Fixed lint issue by handling MetaWalkMut return value; removed unused mapToMessage function
internal/logger_test.go	Comprehensive test suite (21 tests) covering all logging methods, metrics operations, health transitions, and concurrent access with race detection

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

internal/logger.go

internal/stream_module.go

internal/metrics.go

intel352 · 2026-02-23T10:29:37Z

@copilot open a new pull request to apply changes based on the comments in this thread

Copilot · 2026-02-23T10:29:47Z

@intel352 I've opened a new pull request, #9, to work on those changes. Once the pull request is ready, I'll request review from you.

Copilot

Pull request overview

Copilot reviewed 10 out of 10 changed files in this pull request and generated 7 comments.

Comments suppressed due to low confidence (1)

internal/broker_module.go:110

ensureStream runs the per-topic stream with the caller-provided ctx. For a cached, long-lived per-topic stream, using a request-scoped context would stop the stream as soon as that context is canceled. Consider deriving the stream context from the broker/module lifecycle (or context.Background() plus broker shutdown) rather than the call-site ctx.

// ensureStream returns (creating if necessary) a running stream for topic.
// This is used internally when the broker needs a dedicated in-process pipe.
//
//nolint:unused // Reserved for future use by broker consumers.
func (m *brokerModule) ensureStream(ctx context.Context, topic string) (*service.Stream, error) {
	m.mu.RLock()
	if s, ok := m.streams[topic]; ok {
		m.mu.RUnlock()
		return s, nil
	}

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

internal/broker_module.go

internal/metrics.go

internal/output_module.go

internal/stream_module.go

internal/input_module.go

internal/output_module.go

intel352 · 2026-02-23T14:29:32Z

@copilot open a new pull request to apply changes based on the comments in this thread

Copilot · 2026-02-23T14:29:41Z

@intel352 I've opened a new pull request, #13, to work on those changes. Once the pull request is ready, I'll request review from you.

Copilot

Pull request overview

Copilot reviewed 10 out of 10 changed files in this pull request and generated 5 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

internal/output_module.go

internal/stream_module.go

internal/input_module.go

internal/output_module.go

internal/broker_module.go

intel352 · 2026-02-23T20:40:01Z

@copilot open a new pull request to apply changes based on the comments in this thread

and
#7 (review)

Copilot · 2026-02-23T20:40:11Z

@intel352 I've opened a new pull request, #15, to work on those changes. Once the pull request is ready, I'll request review from you.

Copilot

Pull request overview

Copilot reviewed 14 out of 14 changed files in this pull request and generated 6 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

internal/metrics.go

internal/stream_module_test.go

internal/input_module_test.go

internal/output_module_test.go

internal/processor_step.go

internal/logger.go

intel352 · 2026-02-24T00:13:56Z

@copilot open a new pull request to apply changes based on the comments in this thread

Copilot · 2026-02-24T00:14:03Z

@intel352 I've opened a new pull request, #17, to work on those changes. Once the pull request is ready, I'll request review from you.

- Add internal/logger.go: bentoLogger wrapping log/slog with structured fields (component, name) and helpers LogStreamStart, LogStreamStop, LogStreamError, LogMessageProcessed, LogTopicEvent, LogProcessingStart, LogProcessingComplete, LogProcessingError - Add internal/metrics.go: thread-safe StreamMetrics using sync/atomic for message-in/out and error counters, plus mutex-guarded start time and last-message-time; Snapshot() for point-in-time read - Add internal/health.go: healthTracker derives HealthStatus (healthy / degraded / unhealthy) from metrics and running state; HealthReport struct for per-stream health reporting - Update stream_module, input_module, output_module, broker_module: attach logger/metrics/health on construction; emit log lines at start, stop, error, and per-message events; expose Health() method - Update processor_step: log processing start, complete, and error events - Fix pre-existing lint issues: check MetaWalkMut return value, remove unused mapToMessage helper, nolint annotation on future-use ensureStream - Add internal/logger_test.go: 21 unit tests covering log output format, metrics counting, health state transitions, and concurrent metric updates Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Initial plan * fix: apply PR review feedback for observability improvements - Fix wrong comment on LogStreamStop: 'uptimeSeconds' -> 'messagesProcessed' - Use slog.Default() in newLogger to respect global slog configuration - Pass snap.Errors (not always-zero snap.MessagesIn) to LogStreamStop in stream_module - Remove unnecessary mutex lock/unlock in newStreamMetrics initialization Co-authored-by: intel352 <77607+intel352@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: intel352 <77607+intel352@users.noreply.github.com>

… uptime semantics, subscribe ordering (#13) * Initial plan * fix: apply review feedback - lifecycle hooks, health on unexpected exit, subscribe ordering Co-authored-by: intel352 <77607+intel352@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: intel352 <77607+intel352@users.noreply.github.com>

…e for all modules (#15) * Initial plan * fix: stop built stream on subscribe error and add Health() tests for all modules Co-authored-by: intel352 <77607+intel352@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: intel352 <77607+intel352@users.noreply.github.com>

…tic health tests (#17) * Initial plan * fix: hot-path atomics, debug guard, goroutine cleanup, deterministic tests Co-authored-by: intel352 <77607+intel352@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: intel352 <77607+intel352@users.noreply.github.com>

Copilot

Pull request overview

Copilot reviewed 14 out of 14 changed files in this pull request and generated 7 comments.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

internal/processor_step.go

internal/bridge.go

internal/input_module.go

internal/output_module.go

internal/broker_module.go

internal/output_module.go

#19, #20) - #4 output_module: reinitialize done channel at Start for restart support - #6 broker_module: copy streams map under lock then release before calling stream.Stop to prevent potential deadlock in Stop() - #8 processor_step: use sync.Once in consumer func so fan-out (processor emitting multiple messages) does not block on the size-1 channel - #9 processor_step: deferred cleanup selects on ctx.Done() when draining streamDone so a cancelled parent context does not cause an unconditional block - #17 input_module: extract real transport type via inputTransportType() helper so LogStreamStart logs "kafka"/"generate"/etc. instead of "bento.input" - #18 output_module: extract real transport type via outputTransportType() helper so LogStreamStart logs the actual Bento output type - #19 bridge: mapToMessage now returns (*service.Message, error); json.Marshal errors are surfaced instead of silently ignored; updated all call sites - #20 tests: replace remaining time.Sleep(50ms) calls with polling loops in stream_module_test, output_module_test, and broker_module_test Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Copilot AI review requested due to automatic review settings February 23, 2026 10:18

Copilot started reviewing on behalf of intel352 February 23, 2026 10:18 View session

Copilot AI reviewed Feb 23, 2026

View reviewed changes

internal/logger.go Outdated Show resolved Hide resolved

internal/logger.go Show resolved Hide resolved

internal/stream_module.go Outdated Show resolved Hide resolved

internal/metrics.go Show resolved Hide resolved

Copilot AI mentioned this pull request Feb 23, 2026

fix: apply observability PR review feedback #9

Merged

Copilot AI review requested due to automatic review settings February 23, 2026 13:53

Copilot started reviewing on behalf of intel352 February 23, 2026 13:53 View session

Copilot AI reviewed Feb 23, 2026

View reviewed changes

intel352 force-pushed the feat/issue-3-structured-logging branch from 8c10fc6 to 8695bee Compare February 23, 2026 14:03

Copilot AI mentioned this pull request Feb 23, 2026

fix: observability lifecycle correctness — health on unexpected exit, uptime semantics, subscribe ordering #13

Merged

Copilot AI review requested due to automatic review settings February 23, 2026 16:37

Copilot started reviewing on behalf of intel352 February 23, 2026 16:38 View session

Copilot AI reviewed Feb 23, 2026

View reviewed changes

internal/output_module.go Show resolved Hide resolved

internal/stream_module.go Show resolved Hide resolved

internal/input_module.go Show resolved Hide resolved

internal/output_module.go Show resolved Hide resolved

internal/broker_module.go Show resolved Hide resolved

Copilot AI mentioned this pull request Feb 23, 2026

fix: output stream cleanup on subscribe error + Health() test coverage for all modules #15

Merged

Copilot AI review requested due to automatic review settings February 23, 2026 23:59

Copilot started reviewing on behalf of intel352 February 23, 2026 23:59 View session

Copilot AI reviewed Feb 24, 2026

View reviewed changes

Copilot AI mentioned this pull request Feb 24, 2026

fix: hot-path atomics, debug log guard, goroutine cleanup, deterministic health tests #17

Merged

intel352 and others added 5 commits February 23, 2026 19:44

fix: format bridge.go

1ac99b4

chore: retrigger CI

4bda70e

Copilot AI and others added 2 commits February 23, 2026 19:44

Copilot AI review requested due to automatic review settings February 24, 2026 00:46

intel352 force-pushed the feat/issue-3-structured-logging branch from e666723 to dc4bda1 Compare February 24, 2026 00:46

Copilot started reviewing on behalf of intel352 February 24, 2026 00:46 View session

Copilot AI reviewed Feb 24, 2026

View reviewed changes

intel352 merged commit 11b2b9c into main Feb 24, 2026
8 checks passed

intel352 deleted the feat/issue-3-structured-logging branch February 24, 2026 01:37

Conversation

intel352 commented Feb 23, 2026

Summary

Test plan

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

intel352 commented Feb 23, 2026

Uh oh!

Copilot AI commented Feb 23, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

intel352 commented Feb 23, 2026

Uh oh!

Copilot AI commented Feb 23, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

intel352 commented Feb 23, 2026

Uh oh!

Copilot AI commented Feb 23, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

intel352 commented Feb 24, 2026

Uh oh!

Copilot AI commented Feb 24, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants