issue #31 : enable configuring stats emitter in registry

go.mod

@@ -2,6 +2,7 @@ module github.com/modprox/mp
 
 require (
 	github.com/boltdb/bolt v1.3.1
+	github.com/cactus/go-statsd-client v3.1.1+incompatible
 	github.com/go-sql-driver/mysql v1.4.0
 	github.com/gorilla/csrf v1.5.1
 	github.com/gorilla/mux v1.6.2

hack/configs/registry-local.postgres.json

@@ -19,5 +19,9 @@
       "address": "localhost:5432",
       "database": "modproxdb"
     }
+  },
+  "statsd_emitter": {
+    "address": "127.0.0.1",
+    "port": 8125
   }
 }

hack/docker-compose.yaml

@@ -36,3 +36,19 @@ services:
       - POSTGRES_USER=docker
       - POSTGRES_PASSWORD=docker
       - POSTGRES_DB=modproxdb
+
+# A fake datadog/statsd collector which prints metrics
+# to standard out in your docker container
+  fakeadog:
+    image: johnstcn/fakeadog:latest
+    container_name: modprox-fakeadog
+    ports:
+      - target: 8125
+        published: 8125
+        protocol: udp
+        mode: host
+    tmpfs:
+      - /tmp
+    environment:
+      - HOST=0.0.0.0
+      - PORT=8125

registry/config/config.go

@@ -2,13 +2,15 @@ package config
 
 import (
 	"github.com/modprox/mp/pkg/configutil"
+	"github.com/modprox/mp/pkg/netservice"
 	"github.com/pkg/errors"
 )
 
 type Configuration struct {
-	WebServer WebServer       `json:"web_server"`
-	CSRF      CSRF            `json:"csrf"`
-	Database  PersistentStore `json:"database_storage"`
+	WebServer    WebServer       `json:"web_server"`
+	CSRF         CSRF            `json:"csrf"`
+	Database     PersistentStore `json:"database_storage"`
+	StatsEmitter Statsd          `json:"statsd_emitter"`
 }
 
 func (c Configuration) String() string {
@@ -95,3 +97,5 @@ func (dsn DSN) equal(other DSN) bool {
 	}
 	return true
 }
+
+type Statsd netservice.Instance

registry/internal/service/init.go

@@ -5,6 +5,7 @@ import (
 	"net/http"
 	"os"
 
+	"github.com/cactus/go-statsd-client/statsd"
 	"github.com/pkg/errors"
 
 	"github.com/modprox/mp/registry/internal/data"
@@ -13,6 +14,20 @@ import (
 
 type initer func(*Registry) error
 
+func initStatter(r *Registry) error {
+	var err error
+	instance := r.config.StatsEmitter
+	if instance.Port == 0 || instance.Address == "" {
+		r.statter, err = statsd.NewNoopClient()
+		r.log.Warnf("statsd statter is set to noop client")
+		return err
+	}
+	address := fmt.Sprintf("%s:%d", instance.Address, instance.Port)
+	r.statter, err = statsd.NewClient(address, "modprox-registry")
+	r.log.Infof("statsd statter is set to %s", address)
+	return err
+}
+
 func initStore(r *Registry) error {
 	kind, dsn, err := r.config.Database.DSN()
 	if err != nil {
@@ -25,7 +40,7 @@ func initStore(r *Registry) error {
 }
 
 func initWebServer(r *Registry) error {
-	router := web.NewRouter(r.store, r.config.CSRF)
+	router := web.NewRouter(r.store, r.config.CSRF, r.statter)
 
 	listenOn := fmt.Sprintf(
 		"%s:%d",

registry/internal/service/registry.go

@@ -3,15 +3,18 @@ package service
 import (
 	"os"
 
+	"github.com/cactus/go-statsd-client/statsd"
+
 	"github.com/modprox/mp/pkg/loggy"
 	"github.com/modprox/mp/registry/config"
 	"github.com/modprox/mp/registry/internal/data"
 )
 
 type Registry struct {
-	config config.Configuration
-	store  data.Store
-	log    loggy.Logger
+	config  config.Configuration
+	store   data.Store
+	statter statsd.Statter
+	log     loggy.Logger
 }
 
 func NewRegistry(config config.Configuration) *Registry {
@@ -21,6 +24,7 @@ func NewRegistry(config config.Configuration) *Registry {
 	}
 
 	for _, f := range []initer{
+		initStatter,
 		initStore,
 		initWebServer,
 	} {

registry/internal/web/router.go

@@ -3,6 +3,7 @@ package web
 import (
 	"net/http"
 
+	"github.com/cactus/go-statsd-client/statsd"
 	"github.com/gorilla/csrf"
 	"github.com/gorilla/mux"
 	"github.com/shoenig/petrify/v4"
@@ -20,6 +21,7 @@ const (
 func NewRouter(
 	store data.Store,
 	csrfConfig config.CSRF,
+	statter statsd.Statter,
 ) http.Handler {
 
 	// 1) a router onto which sub-routers will be mounted
@@ -34,7 +36,7 @@ func NewRouter(
 	})))
 
 	// 3) an API handler, not CSRF protected
-	router.Handle("/v1/", routeAPI(store))
+	router.Handle("/v1/", routeAPI(store, statter))
 
 	// 4) a webUI handler, is CSRF protected
 	router.Handle("/", routeWebUI(csrfConfig, store))
@@ -49,11 +51,11 @@ func routeStatics(files http.Handler) http.Handler {
 	return sub
 }
 
-func routeAPI(store data.Store) http.Handler {
+func routeAPI(store data.Store, emitter statsd.Statter) http.Handler {
 	sub := mux.NewRouter()
 	sub.Handle("/v1/registry/sources/list", newRegistryList(store)).Methods(get, post)
 	sub.Handle("/v1/registry/sources/new", registryAdd(store)).Methods(post)
-	sub.Handle("/v1/proxy/heartbeat", newHeartbeatHandler(store)).Methods(post)
+	sub.Handle("/v1/proxy/heartbeat", newHeartbeatHandler(store, emitter)).Methods(post)
 	sub.Handle("/v1/proxy/configuration", newStartupHandler(store)).Methods(post)
 	return sub
 }

registry/internal/web/v1_heartbeat.go

@@ -5,6 +5,8 @@ import (
 	"errors"
 	"net/http"
 
+	"github.com/cactus/go-statsd-client/statsd"
+
 	"github.com/modprox/mp/pkg/clients/payloads"
 	"github.com/modprox/mp/pkg/loggy"
 	"github.com/modprox/mp/pkg/netservice"
@@ -13,30 +15,32 @@ import (
 )
 
 type heartbeatHandler struct {
-	store data.Store
-	log   loggy.Logger
+	store   data.Store
+	statter statsd.Statter
+	log     loggy.Logger
 }
 
-func newHeartbeatHandler(store data.Store) http.Handler {
+func newHeartbeatHandler(store data.Store, statter statsd.Statter) http.Handler {
 	return &heartbeatHandler{
-		store: store,
-		log:   loggy.New("heartbeat-update-handler"),
+		store:   store,
+		statter: statter,
+		log:     loggy.New("heartbeat-update-handler"),
 	}
 }
 
 func (h *heartbeatHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
-	h.log.Tracef("receiving heartbeat update from proxy")
-
 	code, response, from, err := h.post(r)
 
 	if err != nil {
 		h.log.Errorf("failed to accept heartbeat from %s, %v", from, err)
 		http.Error(w, response, code)
+		h.statter.Inc("heartbeat-unaccepted", 1, 1)
 		return
 	}
 
 	h.log.Tracef("accepted heartbeat from %s", from)
 	webutil.WriteJSON(w, response)
+	h.statter.Inc("heartbeat-accepted", 1, 1)
 }
 
 func (h *heartbeatHandler) post(r *http.Request) (int, string, netservice.Instance, error) {