diff --git a/docs/user_guide/outputs/tcp_output.md b/docs/user_guide/outputs/tcp_output.md index f0494d00..b086ec55 100644 --- a/docs/user_guide/outputs/tcp_output.md +++ b/docs/user_guide/outputs/tcp_output.md @@ -44,10 +44,20 @@ outputs: keep-alive: # time duration to wait before re-dial in case there is a failure retry-interval: - # NOT IMPLEMENTED boolean, enables the collection and export (via prometheus) of output specific metricss + # maximum number of retries for a pending message after its initial failed delivery attempt. + # defaults to 3. Once exhausted, the message is dropped so the worker can continue. + max-retries: 3 + # boolean, enables the collection and export (via prometheus) of output-specific metrics enable-metrics: false # list of processors to apply on the message before writing event-processors: ``` -A TCP output can be used to export data to an ELK stack, using [Logstash TCP input](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-tcp.html) \ No newline at end of file +When `enable-metrics` is set to `true`, the TCP output exposes: + +- `gnmic_tcp_output_errors_total{name,reason}` for delivery errors. The `reason` label is `dial` or `write`. +- `gnmic_tcp_output_dropped_messages_total{name,reason}` for messages dropped after the retry budget is exhausted. + +`max-retries` applies only after a message has been dequeued for delivery. Dial failures while a message is pending count against its retry budget. + +A TCP output can be used to export data to an ELK stack, using [Logstash TCP input](https://www.elastic.co/guide/en/logstash/current/plugins-inputs-tcp.html) diff --git a/pkg/outputs/tcp_output/tcp_output.go b/pkg/outputs/tcp_output/tcp_output.go index 1b34925d..6dd21bb5 100644 --- a/pkg/outputs/tcp_output/tcp_output.go +++ b/pkg/outputs/tcp_output/tcp_output.go @@ -13,6 +13,7 @@ import ( "encoding/json" "errors" "fmt" + "io" "log/slog" "net" "slices" @@ -21,6 +22,7 @@ import ( "text/template" "time" + "github.com/prometheus/client_golang/prometheus" "google.golang.org/protobuf/proto" "github.com/openconfig/gnmic/pkg/formatters" @@ -34,6 +36,7 @@ import ( const ( defaultRetryTimer = 2 * time.Second defaultNumWorkers = 1 + defaultMaxRetries = 3 outputType = "tcp" ) @@ -53,6 +56,8 @@ type tcpOutput struct { wg *sync.WaitGroup buffer *atomic.Pointer[chan []byte] logger *slog.Logger + name string + reg *prometheus.Registry store store.Store[any] } @@ -77,6 +82,7 @@ type config struct { Delimiter string `mapstructure:"delimiter,omitempty"` KeepAlive time.Duration `mapstructure:"keep-alive,omitempty"` RetryInterval time.Duration `mapstructure:"retry-interval,omitempty"` + MaxRetries int `mapstructure:"max-retries,omitempty"` NumWorkers int `mapstructure:"num-workers,omitempty"` EnableMetrics bool `mapstructure:"enable-metrics,omitempty"` EventProcessors []string `mapstructure:"event-processors,omitempty"` @@ -126,6 +132,8 @@ func (t *tcpOutput) Init(ctx context.Context, name string, cfg map[string]interf } t.store = options.Store + t.name = name + t.reg = options.Registry t.logger = outputs.BindLogger(options.Logger, outputType, name) @@ -153,6 +161,9 @@ func (t *tcpOutput) Init(ctx context.Context, name string, cfg map[string]interf if err != nil { return fmt.Errorf("wrong address format: %v", err) } + if err := t.registerMetrics(newCfg); err != nil { + return err + } ch := make(chan []byte, newCfg.BufferSize) t.buffer.Store(&ch) if newCfg.Rate > 0 { @@ -180,9 +191,15 @@ func setDefaultsFor(cfg *config) { if cfg.NumWorkers < 1 { cfg.NumWorkers = defaultNumWorkers } + if cfg.MaxRetries == 0 { + cfg.MaxRetries = defaultMaxRetries + } } func validate(cfg *config) error { + if cfg.MaxRetries < 0 { + return errors.New("max-retries must be non-negative") + } if cfg.Address == "" { return errors.New("address is required") } @@ -215,6 +232,12 @@ func (t *tcpOutput) Update(_ context.Context, cfg map[string]any) error { setDefaultsFor(newCfg) currCfg := t.cfg.Load() + if newCfg.EnableMetrics && (currCfg == nil || !currCfg.EnableMetrics) { + if err := t.registerMetrics(newCfg); err != nil { + return err + } + } + swapChannel := channelNeedsSwap(currCfg, newCfg) restartWorkers := needsWorkerRestart(currCfg, newCfg) rebuildProcessors := slices.Compare(currCfg.EventProcessors, newCfg.EventProcessors) != 0 @@ -389,53 +412,199 @@ func (t *tcpOutput) String() string { return string(b) } -func (t *tcpOutput) start(ctx context.Context, wg *sync.WaitGroup, idx int) { - defer wg.Done() - workerLogPrefix := fmt.Sprintf("worker-%d", idx) -START: - if ctx.Err() != nil { - t.logger.Warn("context error", "err", ctx.Err()) - return +type tcpDialFunc func(context.Context, string) (net.Conn, error) + +func tcpDialContext(ctx context.Context, address string) (net.Conn, error) { + // Keep keepalive disabled unless the output configuration enables it. + dialer := net.Dialer{KeepAlive: -1} + return dialer.DialContext(ctx, "tcp", address) +} + +func waitTCPRetry(ctx context.Context, interval time.Duration) bool { + timer := time.NewTimer(interval) + defer timer.Stop() + + select { + case <-ctx.Done(): + return false + case <-timer.C: + return true } - cfg := t.cfg.Load() - dc := t.dynCfg.Load() - tcpAddr, err := net.ResolveTCPAddr("tcp", cfg.Address) - if err != nil { - t.logger.Error("failed to resolve address", "worker", workerLogPrefix, "err", err) - time.Sleep(cfg.RetryInterval) - goto START +} + +func writeTCPPayload(w io.Writer, payload []byte) error { + for len(payload) > 0 { + n, err := w.Write(payload) + if n > 0 { + payload = payload[n:] + } + if err != nil { + return err + } + if n == 0 { + return io.ErrNoProgress + } } - conn, err := net.DialTCP("tcp", nil, tcpAddr) - if err != nil { - t.logger.Error("failed to dial TCP", "worker", workerLogPrefix, "err", err) - time.Sleep(cfg.RetryInterval) - goto START + return nil +} + +func (t *tcpOutput) recordTCPError(cfg *config, reason string) { + if cfg.EnableMetrics { + tcpOutputErrors.WithLabelValues(t.name, reason).Inc() } - defer conn.Close() - if cfg.KeepAlive > 0 { - conn.SetKeepAlive(true) - conn.SetKeepAlivePeriod(cfg.KeepAlive) +} + +func (t *tcpOutput) pendingRetryExhausted( + cfg *config, + worker string, + reason string, + retries *int, +) bool { + (*retries)++ + if *retries <= cfg.MaxRetries { + return false + } + + t.logger.Error( + "dropping TCP message after retry limit", + "worker", + worker, + "attempts", + *retries, + "max-retries", + cfg.MaxRetries, + "reason", + reason, + ) + if cfg.EnableMetrics { + tcpOutputDroppedMessages.WithLabelValues(t.name, "max_retries").Inc() } + return true +} + +func (t *tcpOutput) start(ctx context.Context, wg *sync.WaitGroup, idx int) { + t.startWithDialer(ctx, wg, idx, tcpDialContext) +} + +func (t *tcpOutput) startWithDialer( + ctx context.Context, + wg *sync.WaitGroup, + idx int, + dial tcpDialFunc, +) { + defer wg.Done() + + workerLogPrefix := fmt.Sprintf("worker-%d", idx) buffer := *t.buffer.Load() + + var ( + conn net.Conn + pending []byte + pendingRetries int + ) + defer func() { + if conn != nil { + _ = conn.Close() + } + }() + for { - select { - case <-ctx.Done(): + if ctx.Err() != nil { return - case b := <-buffer: - delimiter := dc.delimiter - if dc.limiter != nil { - <-dc.limiter.C - } - // append delimiter - b = append(b, delimiter...) - _, err = conn.Write(b) + } + + cfg := t.cfg.Load() + if conn == nil { + var err error + conn, err = dial(ctx, cfg.Address) if err != nil { - t.logger.Error("failed sending tcp bytes", "worker", workerLogPrefix, "err", err) - conn.Close() - time.Sleep(cfg.RetryInterval) - goto START + if ctx.Err() != nil { + return + } + t.logger.Error( + "failed to dial TCP", + "worker", + workerLogPrefix, + "err", + err, + ) + t.recordTCPError(cfg, "dial") + if pending != nil && t.pendingRetryExhausted( + cfg, + workerLogPrefix, + "dial", + &pendingRetries, + ) { + pending = nil + pendingRetries = 0 + } + if !waitTCPRetry(ctx, cfg.RetryInterval) { + return + } + continue + } + + if tcpConn, ok := conn.(*net.TCPConn); ok && cfg.KeepAlive > 0 { + _ = tcpConn.SetKeepAlive(true) + _ = tcpConn.SetKeepAlivePeriod(cfg.KeepAlive) } } + + if pending == nil { + select { + case <-ctx.Done(): + return + case b := <-buffer: + dc := t.dynCfg.Load() + if dc.limiter != nil { + select { + case <-ctx.Done(): + return + case <-dc.limiter.C: + } + } + + pending = make( + []byte, + 0, + len(b)+len(dc.delimiter), + ) + pending = append(pending, b...) + pending = append(pending, dc.delimiter...) + pendingRetries = 0 + } + } + + if err := writeTCPPayload(conn, pending); err != nil { + t.logger.Error( + "failed sending tcp bytes", + "worker", + workerLogPrefix, + "err", + err, + ) + t.recordTCPError(cfg, "write") + + _ = conn.Close() + conn = nil + + if t.pendingRetryExhausted( + cfg, + workerLogPrefix, + "write", + &pendingRetries, + ) { + pending = nil + pendingRetries = 0 + } + if !waitTCPRetry(ctx, cfg.RetryInterval) { + return + } + continue + } + + pending = nil + pendingRetries = 0 } } diff --git a/pkg/outputs/tcp_output/tcp_output_metrics.go b/pkg/outputs/tcp_output/tcp_output_metrics.go new file mode 100644 index 00000000..7321432b --- /dev/null +++ b/pkg/outputs/tcp_output/tcp_output_metrics.go @@ -0,0 +1,73 @@ +// © 2026 Nokia. +// +// This code is a Contribution to the gNMIc project (“Work”) made under the Google Software Grant and Corporate Contributor License Agreement (“CLA”) and governed by the Apache License 2.0. +// No other rights or licenses in or to any of Nokia’s intellectual property are granted for any other purpose. +// This code is provided on an “as is” basis without any warranties of any kind. +// +// SPDX-License-Identifier: Apache-2.0 + +package tcp_output + +import ( + "errors" + + "github.com/prometheus/client_golang/prometheus" +) + +var tcpOutputErrors = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "gnmic", + Subsystem: "tcp_output", + Name: "errors_total", + Help: "Number of TCP output delivery errors", + }, + []string{"name", "reason"}, +) + +var tcpOutputDroppedMessages = prometheus.NewCounterVec( + prometheus.CounterOpts{ + Namespace: "gnmic", + Subsystem: "tcp_output", + Name: "dropped_messages_total", + Help: "Number of TCP output messages dropped", + }, + []string{"name", "reason"}, +) + +func registerTCPCollector( + reg *prometheus.Registry, + collector prometheus.Collector, +) error { + if err := reg.Register(collector); err != nil { + var alreadyRegistered prometheus.AlreadyRegisteredError + if errors.As(err, &alreadyRegistered) { + return nil + } + return err + } + return nil +} + +func (t *tcpOutput) registerMetrics(cfg *config) error { + if cfg == nil || !cfg.EnableMetrics { + return nil + } + if t.reg == nil { + t.logger.Error( + "metrics enabled but main registry is not initialized, enable main metrics under `api-server`", + ) + return nil + } + + if err := registerTCPCollector(t.reg, tcpOutputErrors); err != nil { + return err + } + if err := registerTCPCollector(t.reg, tcpOutputDroppedMessages); err != nil { + return err + } + + tcpOutputErrors.WithLabelValues(t.name, "dial").Add(0) + tcpOutputErrors.WithLabelValues(t.name, "write").Add(0) + tcpOutputDroppedMessages.WithLabelValues(t.name, "max_retries").Add(0) + return nil +} diff --git a/pkg/outputs/tcp_output/tcp_output_test.go b/pkg/outputs/tcp_output/tcp_output_test.go index f6b22ca9..837c9b39 100644 --- a/pkg/outputs/tcp_output/tcp_output_test.go +++ b/pkg/outputs/tcp_output/tcp_output_test.go @@ -9,16 +9,73 @@ package tcp_output import ( + "bytes" "context" + "errors" + "io" "net" "strings" + "sync" + "sync/atomic" "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" "github.com/openconfig/gnmic/pkg/outputs" "github.com/zestor-dev/zestor/store" "github.com/zestor-dev/zestor/store/gomap" ) +type shortWriter struct { + bytes.Buffer +} + +func (w *shortWriter) Write(p []byte) (int, error) { + if len(p) > 1 { + p = p[:1] + } + return w.Buffer.Write(p) +} + +type partialErrorConn struct { + n int +} + +func (c *partialErrorConn) Read([]byte) (int, error) { + return 0, io.EOF +} + +func (c *partialErrorConn) Write(p []byte) (int, error) { + n := min(c.n, len(p)) + return n, errors.New("forced write failure") +} + +func (*partialErrorConn) Close() error { + return nil +} + +func (*partialErrorConn) LocalAddr() net.Addr { + return nil +} + +func (*partialErrorConn) RemoteAddr() net.Addr { + return nil +} + +func (*partialErrorConn) SetDeadline(time.Time) error { + return nil +} + +func (*partialErrorConn) SetReadDeadline(time.Time) error { + return nil +} + +func (*partialErrorConn) SetWriteDeadline(time.Time) error { + return nil +} + func newStore() store.Store[any] { return gomap.NewMemStore(store.StoreOptions[any]{}) } @@ -58,6 +115,9 @@ func TestTCP_SetDefaults(t *testing.T) { if c.NumWorkers != defaultNumWorkers { t.Errorf("num workers default") } + if c.MaxRetries != defaultMaxRetries { + t.Errorf("max retries default = %d, want %d", c.MaxRetries, defaultMaxRetries) + } } func TestTCP_Validate(t *testing.T) { @@ -82,6 +142,13 @@ func TestTCP_Validate(t *testing.T) { if err := t1.Validate(map[string]any{"buffer-size": "x"}); err == nil { t.Errorf("expected decode error") } + if err := t1.Validate(map[string]any{ + "address": "127.0.0.1:1", + "target-template": "foo", + "max-retries": -1, + }); err == nil { + t.Errorf("expected negative max-retries error") + } } func TestTCP_InitAndUpdate(t *testing.T) { @@ -163,6 +230,273 @@ func TestTCP_InitErrors(t *testing.T) { } } +func TestTCP_WritePayloadHandlesShortWrites(t *testing.T) { + writer := new(shortWriter) + want := []byte("telemetry") + + if err := writeTCPPayload(writer, want); err != nil { + t.Fatalf("writeTCPPayload() error = %v", err) + } + if got := writer.Bytes(); !bytes.Equal(got, want) { + t.Fatalf("writeTCPPayload() wrote %q, want %q", got, want) + } +} + +func TestTCP_RetriesMessageAfterWriteFailure(t *testing.T) { + o := &tcpOutput{} + o.init() + + cfg := &config{ + Address: "127.0.0.1:1", + RetryInterval: time.Millisecond, + MaxRetries: defaultMaxRetries, + } + o.cfg.Store(cfg) + o.dynCfg.Store(&dynConfig{delimiter: []byte("\n")}) + + buffer := make(chan []byte, 2) + o.buffer.Store(&buffer) + + secondClient, secondServer := net.Pipe() + defer secondServer.Close() + + connections := make(chan net.Conn, 2) + connections <- &partialErrorConn{n: 2} + connections <- secondClient + + var dialCount atomic.Int32 + dial := func(ctx context.Context, _ string) (net.Conn, error) { + dialCount.Add(1) + select { + case conn := <-connections: + return conn, nil + case <-ctx.Done(): + return nil, ctx.Err() + } + } + + ctx, cancel := context.WithCancel(context.Background()) + wg := new(sync.WaitGroup) + wg.Add(1) + go o.startWithDialer(ctx, wg, 0, dial) + + buffer <- []byte("first") + buffer <- []byte("second") + + want := []byte("first\nsecond\n") + got := make([]byte, len(want)) + if err := secondServer.SetReadDeadline( + time.Now().Add(5 * time.Second), + ); err != nil { + t.Fatalf("SetReadDeadline() error = %v", err) + } + if _, err := io.ReadFull(secondServer, got); err != nil { + t.Fatalf("ReadFull() error = %v", err) + } + if !bytes.Equal(got, want) { + t.Fatalf("received %q, want %q", got, want) + } + + cancel() + + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("TCP worker did not stop after cancellation") + } + + if got := dialCount.Load(); got != 2 { + t.Fatalf("dial count = %d, want 2", got) + } +} + +func TestTCP_RegisterMetrics(t *testing.T) { + o := &tcpOutput{} + o.init() + o.name = t.Name() + o.reg = prometheus.NewRegistry() + cfg := &config{EnableMetrics: true} + o.cfg.Store(cfg) + + if err := o.registerMetrics(cfg); err != nil { + t.Fatalf("registerMetrics() error = %v", err) + } + + families, err := o.reg.Gather() + if err != nil { + t.Fatalf("Gather() error = %v", err) + } + + got := make(map[string]bool, len(families)) + for _, family := range families { + got[family.GetName()] = true + } + + for _, want := range []string{ + "gnmic_tcp_output_errors_total", + "gnmic_tcp_output_dropped_messages_total", + } { + if !got[want] { + t.Errorf("registered metrics missing %q", want) + } + } +} + +func TestTCP_UpdateEnablesMetrics(t *testing.T) { + addr, stop := freeTCPListener(t) + defer stop() + + o := &tcpOutput{} + reg := prometheus.NewRegistry() + cfg := map[string]any{ + "address": addr, + "format": "json", + "buffer-size": 16, + "num-workers": 1, + "target-template": "{{ .target }}", + "enable-metrics": false, + } + if err := o.Init( + context.Background(), + t.Name(), + cfg, + outputs.WithConfigStore(newStore()), + outputs.WithRegistry(reg), + ); err != nil { + t.Fatalf("Init() error = %v", err) + } + defer o.Close() + + cfg["enable-metrics"] = true + if err := o.Update(context.Background(), cfg); err != nil { + t.Fatalf("Update() error = %v", err) + } + + families, err := reg.Gather() + if err != nil { + t.Fatalf("Gather() error = %v", err) + } + got := make(map[string]bool, len(families)) + for _, family := range families { + got[family.GetName()] = true + } + for _, want := range []string{ + "gnmic_tcp_output_errors_total", + "gnmic_tcp_output_dropped_messages_total", + } { + if !got[want] { + t.Errorf("metrics after Update missing %q", want) + } + } +} + +func TestTCP_DropsMessageAfterRetryLimit(t *testing.T) { + o := &tcpOutput{} + o.init() + o.name = t.Name() + + cfg := &config{ + Address: "127.0.0.1:1", + RetryInterval: time.Millisecond, + MaxRetries: 2, + EnableMetrics: true, + } + o.cfg.Store(cfg) + o.dynCfg.Store(&dynConfig{delimiter: []byte("\n")}) + + buffer := make(chan []byte, 2) + o.buffer.Store(&buffer) + + successClient, successServer := net.Pipe() + defer successServer.Close() + + var dialCount atomic.Int32 + dial := func(ctx context.Context, _ string) (net.Conn, error) { + switch dialCount.Add(1) { + case 1: + return &partialErrorConn{n: 0}, nil + case 2, 3: + return nil, errors.New("forced dial failure") + case 4: + return successClient, nil + default: + <-ctx.Done() + return nil, ctx.Err() + } + } + + writeErrorsBefore := testutil.ToFloat64( + tcpOutputErrors.WithLabelValues(o.name, "write"), + ) + dialErrorsBefore := testutil.ToFloat64( + tcpOutputErrors.WithLabelValues(o.name, "dial"), + ) + droppedBefore := testutil.ToFloat64( + tcpOutputDroppedMessages.WithLabelValues(o.name, "max_retries"), + ) + + ctx, cancel := context.WithCancel(context.Background()) + wg := new(sync.WaitGroup) + wg.Add(1) + go o.startWithDialer(ctx, wg, 0, dial) + + buffer <- []byte("first") + buffer <- []byte("second") + + want := []byte("second\n") + got := make([]byte, len(want)) + if err := successServer.SetReadDeadline( + time.Now().Add(5 * time.Second), + ); err != nil { + t.Fatalf("SetReadDeadline() error = %v", err) + } + if _, err := io.ReadFull(successServer, got); err != nil { + t.Fatalf("ReadFull() error = %v", err) + } + if !bytes.Equal(got, want) { + t.Fatalf("received %q, want %q", got, want) + } + + cancel() + done := make(chan struct{}) + go func() { + wg.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("TCP worker did not stop after cancellation") + } + + if got := dialCount.Load(); got != 4 { + t.Fatalf("dial count = %d, want 4", got) + } + + if got := testutil.ToFloat64( + tcpOutputErrors.WithLabelValues(o.name, "write"), + ) - writeErrorsBefore; got != 1 { + t.Errorf("write error metric delta = %v, want 1", got) + } + if got := testutil.ToFloat64( + tcpOutputErrors.WithLabelValues(o.name, "dial"), + ) - dialErrorsBefore; got != 2 { + t.Errorf("dial error metric delta = %v, want 2", got) + } + if got := testutil.ToFloat64( + tcpOutputDroppedMessages.WithLabelValues(o.name, "max_retries"), + ) - droppedBefore; got != 1 { + t.Errorf("dropped metric delta = %v, want 1", got) + } +} + func TestTCP_Predicates(t *testing.T) { a := &config{BufferSize: 1, NumWorkers: 1} b := &config{BufferSize: 2, NumWorkers: 1}