Merge pull request #1346 from jehiah/statsd_exclude_ephemeral_1346

nsqd: configurable to skip ephemeral topics/channels in statsd output
nsqio · May 21, 2021 · a493996 · a493996
2 parents d0f7412 + a12d747
commit a493996
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 5 deletions.
diff --git a/apps/nsqd/options.go b/apps/nsqd/options.go
@@ -163,6 +163,7 @@ func nsqdFlagSet(opts *nsqd.Options) *flag.FlagSet {
 	flagSet.Bool("statsd-mem-stats", opts.StatsdMemStats, "toggle sending memory and GC stats to statsd")
 	flagSet.String("statsd-prefix", opts.StatsdPrefix, "prefix used for keys sent to statsd (%s for host replacement)")
 	flagSet.Int("statsd-udp-packet-size", opts.StatsdUDPPacketSize, "the size in bytes of statsd UDP packets")
+	flagSet.Bool("statsd-exclude-ephemeral", opts.StatsdExcludeEphemeral, "Skip ephemeral topics and channels when sending stats to statsd")
 
 	// End to end percentile flags
 	e2eProcessingLatencyPercentiles := app.FloatArray{}

diff --git a/nsqd/options.go b/nsqd/options.go
@@ -61,11 +61,12 @@ type Options struct {
 	MaxChannelConsumers    int           `flag:"max-channel-consumers"`
 
 	// statsd integration
-	StatsdAddress       string        `flag:"statsd-address"`
-	StatsdPrefix        string        `flag:"statsd-prefix"`
-	StatsdInterval      time.Duration `flag:"statsd-interval"`
-	StatsdMemStats      bool          `flag:"statsd-mem-stats"`
-	StatsdUDPPacketSize int           `flag:"statsd-udp-packet-size"`
+	StatsdAddress          string        `flag:"statsd-address"`
+	StatsdPrefix           string        `flag:"statsd-prefix"`
+	StatsdInterval         time.Duration `flag:"statsd-interval"`
+	StatsdMemStats         bool          `flag:"statsd-mem-stats"`
+	StatsdUDPPacketSize    int           `flag:"statsd-udp-packet-size"`
+	StatsdExcludeEphemeral bool          `flag:"statsd-exclude-ephemeral"`
 
 	// e2e message latency
 	E2EProcessingLatencyWindowTime  time.Duration `flag:"e2e-processing-latency-window-time"`

diff --git a/nsqd/statsd.go b/nsqd/statsd.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"math"
 	"net"
+	"strings"
 	"time"
 
 	"github.com/nsqio/nsq/internal/statsd"
@@ -36,6 +37,7 @@ func (n *NSQD) statsdLoop() {
 		case <-ticker.C:
 			addr := n.getOpts().StatsdAddress
 			prefix := n.getOpts().StatsdPrefix
+			excludeEphemeral := n.getOpts().StatsdExcludeEphemeral
 			conn, err := net.DialTimeout("udp", addr, time.Second)
 			if err != nil {
 				n.logf(LOG_ERROR, "failed to create UDP socket to statsd(%s)", addr)
@@ -49,6 +51,10 @@ func (n *NSQD) statsdLoop() {
 
 			stats := n.GetStats("", "", false)
 			for _, topic := range stats.Topics {
+				if excludeEphemeral && strings.HasSuffix(topic.TopicName, "#ephemeral") {
+					continue
+				}
+
 				// try to find the topic in the last collection
 				lastTopic := TopicStats{}
 				for _, checkTopic := range lastStats.Topics {
@@ -80,6 +86,10 @@ func (n *NSQD) statsdLoop() {
 				}
 
 				for _, channel := range topic.Channels {
+					if excludeEphemeral && strings.HasSuffix(channel.ChannelName, "#ephemeral") {
+						continue
+					}
+
 					// try to find the channel in the last collection
 					lastChannel := ChannelStats{}
 					for _, checkChannel := range lastTopic.Channels {