Repository: atlassian/gostatsd
Pull request #259 (2019-08-22 06:34:07)

support debug-mode to print custom metrics to stdout.
- Provide a debug-mode configurable item which defaults to false.
- Once the debug-mode is on, the server should print the metrics received from network to stdout.
  - Should ONLY print the custom metrics.
  - If the terminal/stdout speed is less than metrics incoming speed, it’s reasonable to ignore some of them. So you could not see all the metrics sent to the server from the stdout under such circumstances.

Sample Usage
- dummy client
`while [[ 1 ]];do echo 'abc.def.g:10|c' | nc -w1 -u localhost 8125;echo -n '.';sleep 1;done`
- server with --log-raw-metric=true
`{"level":"info","msg":"[{counter, abc.def.g, 10.000000, , }]","time":"2019-08-26T16:04:33+10:00"}`


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 8d2509f..b67acf9 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,3 +1,7 @@
+17.2.0
+------
+- Support `log-raw-metric` configuration
+
 17.1.2
 ------
 - Remove support for disabling retries in the forwarder
diff --git a/README.md b/README.md
index 0903bad..79dad7e 100644
--- a/README.md
+++ b/README.md
@@ -83,7 +83,8 @@ following configuration options:
   configured parsers (`--max-parsers` option).  Defaults to the value of `--max-parsers`, but may require tuning for
   HTTP based servers.
 - `flush-interval`: duration for how long to batch metrics before flushing. Should be an order of magnitude less than
-  the upstream flush interval. Defaults to `1s`
+  the upstream flush interval. Defaults to `1s`.
+- `log-raw-metric`: print metrics received from network to stdout in JSON format.
 
 Configuring HTTP servers
 ------------------------
diff --git a/cmd/gostatsd/main.go b/cmd/gostatsd/main.go
index 3742f8a..609d8e6 100644
--- a/cmd/gostatsd/main.go
+++ b/cmd/gostatsd/main.go
@@ -133,6 +133,7 @@ func constructServer(v *viper.Viper) (*statsd.Server, error) {
 		ReceiveBatchSize:    v.GetInt(statsd.ParamReceiveBatchSize),
 		ConnPerReader:       v.GetBool(statsd.ParamConnPerReader),
 		ServerMode:          v.GetString(statsd.ParamServerMode),
+		LogRawMetric:        v.GetBool(statsd.ParamLogRawMetric),
 		CacheOptions: statsd.CacheOptions{
 			CacheRefreshPeriod:        v.GetDuration(statsd.ParamCacheRefreshPeriod),
 			CacheEvictAfterIdlePeriod: v.GetDuration(statsd.ParamCacheEvictAfterIdlePeriod),
diff --git a/pkg/statsd/parser.go b/pkg/statsd/parser.go
index b2cb74f..1bceab3 100644
--- a/pkg/statsd/parser.go
+++ b/pkg/statsd/parser.go
@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -15,6 +16,9 @@ import (
 	"golang.org/x/time/rate"
 )
 
+// Default buffer size for debug channel
+const logRawMetricChannelBufferSize = 1000
+
 // DatagramParser receives datagrams and parses them into Metrics/Events
 // For each Metric/Event it calls Handler.HandleMetric/Event()
 type DatagramParser struct {
@@ -34,10 +38,14 @@ type DatagramParser struct {
 	badLineLimiter *rate.Limiter
 
 	in <-chan []*Datagram // Input chan of datagram batches to parse
+
+	logRawMetric         bool
+	logRawMetricInitOnce sync.Once
+	logRawMetricChan     chan []*gostatsd.Metric
 }
 
 // NewDatagramParser initialises a new DatagramParser.
-func NewDatagramParser(in <-chan []*Datagram, ns string, ignoreHost bool, estimatedTags int, handler gostatsd.PipelineHandler, badLineRateLimitPerSecond rate.Limit) *DatagramParser {
+func NewDatagramParser(in <-chan []*Datagram, ns string, ignoreHost bool, estimatedTags int, handler gostatsd.PipelineHandler, badLineRateLimitPerSecond rate.Limit, logRawMetric bool) *DatagramParser {
 	limiter := &rate.Limiter{}
 	if badLineRateLimitPerSecond > 0 {
 		limiter = rate.NewLimiter(badLineRateLimitPerSecond, 1)
@@ -50,6 +58,7 @@ func NewDatagramParser(in <-chan []*Datagram, ns string, ignoreHost bool, estima
 		namespace:      ns,
 		metricPool:     pool.NewMetricPool(estimatedTags + handler.EstimatedTags()),
 		badLineLimiter: limiter,
+		logRawMetric:   logRawMetric,
 	}
 }
 
@@ -71,6 +80,8 @@ func (dp *DatagramParser) RunMetrics(ctx context.Context) {
 }
 
 func (dp *DatagramParser) Run(ctx context.Context) {
+	dp.initLogRawMetric(ctx)
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -89,6 +100,8 @@ func (dp *DatagramParser) Run(ctx context.Context) {
 			}
 			if len(metrics) > 0 {
 				dp.handler.DispatchMetrics(ctx, metrics)
+
+				dp.doLogRawMetric(metrics)
 			}
 			atomic.AddUint64(&dp.metricsReceived, uint64(len(metrics)))
 			atomic.AddUint64(&dp.eventsReceived, accumE)
@@ -170,3 +183,37 @@ func (dp *DatagramParser) parseLine(line []byte) (*gostatsd.Metric, *gostatsd.Ev
 	}
 	return l.run(line, dp.namespace)
 }
+
+func (dp *DatagramParser) initLogRawMetric(ctx context.Context) {
+	if dp.logRawMetric {
+		dp.logRawMetricInitOnce.Do(func() {
+			dp.logRawMetricChan = make(chan []*gostatsd.Metric, logRawMetricChannelBufferSize)
+
+			go func() {
+				// The `StandardLogger` returns the singleton log instance which is used elsewhere in the application.
+				// And the `log` package makes sure that the logging interface are thread-safe so that we wouldn't
+				// mess up the contents of stdout/stderr.
+				lg := log.StandardLogger()
+				for {
+					select {
+					case <-ctx.Done():
+						return
+					case metrics := <-dp.logRawMetricChan:
+						lg.Info(metrics)
+					}
+				}
+			}()
+		})
+	}
+}
+
+func (dp *DatagramParser) doLogRawMetric(metrics []*gostatsd.Metric) {
+	if dp.logRawMetric {
+		// Deliver only once, we don't want the whole pipeline being blocked due to a slow terminal.
+		// So may lose packet if the channel buffer is full.
+		select {
+		case dp.logRawMetricChan <- metrics:
+		default:
+		}
+	}
+}
diff --git a/pkg/statsd/statsd.go b/pkg/statsd/statsd.go
index 9031512..331efcf 100644
--- a/pkg/statsd/statsd.go
+++ b/pkg/statsd/statsd.go
@@ -51,6 +51,7 @@ type Server struct {
 	BadLineRateLimitPerSecond rate.Limit
 	ServerMode                string
 	Hostname                  string
+	LogRawMetric              bool
 	CacheOptions
 	Viper *viper.Viper
 }
@@ -168,7 +169,7 @@ func (s *Server) RunWithCustomSocket(ctx context.Context, sf SocketFactory) erro
 	datagrams := make(chan []*Datagram)
 
 	// Create the Parser
-	parser := NewDatagramParser(datagrams, s.Namespace, s.IgnoreHost, s.EstimatedTags, handler, s.BadLineRateLimitPerSecond)
+	parser := NewDatagramParser(datagrams, s.Namespace, s.IgnoreHost, s.EstimatedTags, handler, s.BadLineRateLimitPerSecond, s.LogRawMetric)
 	runnables = append(runnables, parser.RunMetrics)
 	for i := 0; i < s.MaxParsers; i++ {
 		runnables = append(runnables, parser.Run)
diff --git a/pkg/statsd/statsd_defaults_and_params.go b/pkg/statsd/statsd_defaults_and_params.go
index f9ff3b6..4243572 100644
--- a/pkg/statsd/statsd_defaults_and_params.go
+++ b/pkg/statsd/statsd_defaults_and_params.go
@@ -84,6 +84,8 @@ const (
 	DefaultBadLinesPerMinute = 0
 	// DefaultServerMode is the default mode to run as, standalone|forwarder
 	DefaultServerMode = "standalone"
+	// DefaultLogRawMetric is the default value for whether to log the metrics received from network
+	DefaultLogRawMetric = false
 )
 
 const (
@@ -147,6 +149,8 @@ const (
 	ParamServerMode = "server-mode"
 	// ParamHostname allows hostname overrides
 	ParamHostname = "hostname"
+	// ParamLogRawMetric enables custom metrics to be printed to stdout
+	ParamLogRawMetric = "log-raw-metric"
 )
 
 // AddFlags adds flags to the specified FlagSet.
@@ -180,6 +184,7 @@ func AddFlags(fs *pflag.FlagSet) {
 	fs.Bool(ParamConnPerReader, DefaultConnPerReader, "Create a separate connection per reader (requires system support for reusing addresses)")
 	fs.String(ParamServerMode, DefaultServerMode, "The server mode to run in")
 	fs.String(ParamHostname, getHost(), "overrides the hostname of the server")
+	fs.Bool(ParamLogRawMetric, DefaultLogRawMetric, "Print metrics received from network to stdout in JSON format")
 }
 
 func minInt(a, b int) int {


Test patch:
diff --git a/pkg/statsd/parser_test.go b/pkg/statsd/parser_test.go
index cd6cfac..d3ca5f4 100644
--- a/pkg/statsd/parser_test.go
+++ b/pkg/statsd/parser_test.go
@@ -20,7 +20,7 @@ const fakeIP = gostatsd.IP("127.0.0.1")
 
 func newTestParser(ignoreHost bool) (*DatagramParser, *countingHandler) {
 	ch := &countingHandler{}
-	return NewDatagramParser(nil, "", ignoreHost, 0, ch, rate.Limit(0)), ch
+	return NewDatagramParser(nil, "", ignoreHost, 0, ch, rate.Limit(0), false), ch
 }
 
 func TestParseEmptyDatagram(t *testing.T) {