eventstat: publisher to send out telemetry compatible UDP packages

Biggest part of this patch is just moving code out from telemetry/client.go to it them reusable.

Other important part is the eventstat/publisher.go which re-uses the existing code from
telemetry/client.go to send out UDP packages in the same format.

Change-Id: I68be36017bcb4ef010bacf8b8d843d541150ded9

Author: elek

eventstat/publisher.go

@@ -0,0 +1,124 @@
+// Copyright (C) 2022 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package eventstat
+
+import (
+	"context"
+	"os"
+	"strings"
+	"time"
+
+	"storj.io/common/telemetry"
+)
+
+// ClientOpts allows you to set Client Options.
+type ClientOpts struct {
+	// Interval is how frequently stats from the provided Registry will be
+	// sent up. Note that this interval is "jittered", so the actual interval
+	// is taken from a normal distribution with a mean of Interval and a
+	// variance of Interval/4. Defaults to DefaultInterval.
+	Interval time.Duration
+
+	// Application is the application name, usually prepended to metric names.
+	// By default it will be os.Args[0].
+	Application string
+
+	// Instance is a string that identifies this particular server. Could be a
+	// node id, but defaults to the result of DefaultInstanceId().
+	Instance string
+
+	// PacketSize controls how we fragment the data as it goes out in UDP
+	// packets. Defaults to DefaultPacketSize.
+	PacketSize int
+}
+
+// UDPPublisher is an  eventstat telemetry client for sending UDP packets at a regular interval.
+type UDPPublisher struct {
+	reporter *telemetry.Reporter
+}
+
+// NewUDPPublisher constructs a telemetry client that sends packets to remoteAddr
+// over UDP.
+func NewUDPPublisher(remoteAddr string, registry *Registry, opts ClientOpts) (rv *UDPPublisher, err error) {
+	if opts.Interval == 0 {
+		opts.Interval = telemetry.DefaultInterval
+	}
+	if opts.Application == "" {
+		if len(os.Args) > 0 {
+			opts.Application = os.Args[0]
+		} else {
+			// what the actual heck
+			opts.Application = telemetry.DefaultApplication
+		}
+	}
+	if opts.Instance == "" {
+		opts.Instance = telemetry.DefaultInstanceID()
+	}
+	if opts.PacketSize == 0 {
+		opts.PacketSize = telemetry.DefaultPacketSize
+	}
+
+	udpOptions := telemetry.Options{
+		Application: opts.Application,
+		InstanceID:  []byte(opts.Instance),
+		Address:     remoteAddr,
+		PacketSize:  opts.PacketSize,
+	}
+	reporter, err := telemetry.NewReporter(opts.Interval, func(ctx context.Context) error {
+		return telemetry.Send(ctx, udpOptions, func(publishEntry func(key string, value float64)) {
+			registry.PublishAndReset(func(name string, tags Tags, value float64) {
+				telemetryKey := telemetryKey(name, tags)
+				publishEntry(telemetryKey, value)
+			})
+		})
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &UDPPublisher{
+		reporter: reporter,
+	}, nil
+}
+
+func telemetryKey(name string, tags Tags) string {
+	builder := strings.Builder{}
+	writeTag(&builder, name)
+	if len(tags) > 0 {
+		builder.WriteString(",")
+		builder.WriteString(tags.String())
+	}
+	builder.WriteString(" value")
+	telemetryKey := builder.String()
+	return telemetryKey
+}
+
+// Run calls Report roughly every Interval.
+func (c *UDPPublisher) Run(ctx context.Context) {
+	c.reporter.Run(ctx)
+}
+
+// publish sends out message immediately independent on Interval.
+func (c *UDPPublisher) publish(ctx context.Context) error {
+	return c.reporter.Publish(ctx)
+}
+
+// writeTag writes a tag key, value, or field key to the builder.
+func writeTag(builder *strings.Builder, tag string) {
+	if strings.IndexByte(tag, ',') == -1 &&
+		strings.IndexByte(tag, '=') == -1 &&
+		strings.IndexByte(tag, ' ') == -1 {
+
+		builder.WriteString(tag)
+		return
+	}
+
+	for i := 0; i < len(tag); i++ {
+		if tag[i] == ',' ||
+			tag[i] == '=' ||
+			tag[i] == ' ' {
+			builder.WriteByte('\\')
+		}
+		builder.WriteByte(tag[i])
+	}
+}

eventstat/publisher_test.go

@@ -0,0 +1,94 @@
+// Copyright (C) 2022 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package eventstat
+
+import (
+	"fmt"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"storj.io/common/telemetry"
+	"storj.io/common/testcontext"
+)
+
+func TestMetrics(t *testing.T) {
+	ctx := testcontext.New(t)
+
+	if runtime.GOOS == "windows" {
+		// TODO(windows): currently closing doesn't seem to be shutting down the server
+		t.Skip("broken")
+	}
+
+	s, err := telemetry.Listen("127.0.0.1:0")
+	assert.NoError(t, err)
+	defer func() { _ = s.Close() }()
+
+	r := Registry{}
+	counter := r.NewTagCounter("http_user_agent", "agent")
+	fmt.Println(s.Addr())
+	c, err := NewUDPPublisher(s.Addr(), &r, ClientOpts{
+		Application: "testapp",
+		Instance:    "testinst",
+		Interval:    10 * time.Millisecond,
+	})
+	require.NoError(t, err)
+
+	counter("aws")
+	counter("aws")
+	counter("aws")
+	counter("rclone")
+
+	err = c.publish(ctx)
+	require.NoError(t, err)
+
+	expectedMetric := 4
+
+	keys := make(chan string, expectedMetric)
+	values := make(chan float64, expectedMetric)
+	defer close(keys)
+	defer close(values)
+
+	ctx.Go(func() error {
+		fmt.Println("Listening on " + s.Addr())
+		// note: this is the telemetry server which guarantees that our sender is still compatible with the format
+		_ = s.Serve(ctx, telemetry.HandlerFunc(
+			func(application, instance string, key []byte, val float64) {
+				assert.Equal(t, application, "testapp")
+				assert.Equal(t, instance, "testinst")
+				keys <- string(key)
+				values <- val
+			}))
+		return nil
+	})
+
+	for i := 0; i < expectedMetric; i++ {
+		key := <-keys
+		value := <-values
+
+		switch key {
+		case "http_user_agent_count,agent=aws value":
+			assert.Equal(t, float64(3), value)
+		case "http_user_agent_count,agent=rclone value":
+			assert.Equal(t, float64(1), value)
+		case "http_user_agent_discarded value":
+			assert.Equal(t, float64(0), value)
+		case "http_user_agent_buckets value":
+			assert.Equal(t, float64(2), value)
+		default:
+			require.Failf(t, "Unexpected UDP metric", "key=%s", key)
+		}
+
+	}
+
+}
+
+func TestTelemetryKey(t *testing.T) {
+	assert.Equal(t, "key1 value", telemetryKey("key1", Tags{}))
+	assert.Equal(t, "key2,foo=bar value", telemetryKey("key2", Tags{"foo": "bar"}))
+	assert.Equal(t, "key3,f\\==ba\\,r value", telemetryKey("key3", Tags{"f=": "ba,r"}))
+}

eventstat/registry.go

@@ -4,12 +4,27 @@
 package eventstat
 
 import (
+	"strings"
 	"sync"
 )
 
 // Tags represent key/values for any event.
 type Tags map[string]string
 
+func (t *Tags) String() string {
+	var builder strings.Builder
+	for k, v := range *t {
+		if builder.Len() > 0 {
+			builder.WriteString(",")
+		}
+		writeTag(&builder, k)
+		builder.WriteString("=")
+		writeTag(&builder, v)
+
+	}
+	return builder.String()
+}
+
 // Publisher is a function which sends out statistics.
 type Publisher func(name string, tags Tags, value float64)
 

eventstat/registry_test.go

@@ -75,6 +75,13 @@ func TestRegistry_WithLimit(t *testing.T) {
 	}, p.sortedEvents())
 }
 
+func TestTags(t *testing.T) {
+	require.Equal(t, "foo=bar", (&eventstat.Tags{"foo": "bar"}).String())
+	twoKeys := (&eventstat.Tags{"foo1": "bar", "foo2": "bar"}).String()
+	require.True(t, twoKeys == "foo1=bar,foo2=bar" || twoKeys == "foo2=bar,foo1=bar")
+	require.Equal(t, "f\\=oo=b\\,a\\ r", (&eventstat.Tags{"f=oo": "b,a r"}).String())
+}
+
 func BenchmarkRegistry(b *testing.B) {
 	b.ReportAllocs()
 	r := eventstat.Registry{}