Add health check listener, closes #42

Author: Le Cheng Fan

cmd/trade.go

@@ -11,6 +11,7 @@ import (
 	"github.com/lightyeario/kelp/model"
 	"github.com/lightyeario/kelp/plugins"
 	"github.com/lightyeario/kelp/support/monitoring"
+	"github.com/lightyeario/kelp/support/networking"
 	"github.com/lightyeario/kelp/support/utils"
 	"github.com/lightyeario/kelp/trader"
 	"github.com/spf13/cobra"
@@ -119,6 +120,7 @@ func init() {
 			// we want to delete all the offers and exit here since there is something wrong with our setup
 			deleteAllOffersAndExit(botConfig, client, sdex)
 		}
+
 		bot := trader.MakeBot(
 			client,
 			botConfig.AssetBase(),
@@ -136,6 +138,22 @@ func init() {
 		validateTrustlines(client, &botConfig)
 		log.Printf("trustlines valid\n")
 
+		// --- start initialization of services ---
+		if botConfig.MonitoringPort != 0 {
+			go func() {
+				e := startMonitoringServer(botConfig)
+				if e != nil {
+					log.Println()
+					log.Printf("unable to start the monitoring server or problem encountered while running server: %s\n", e)
+					// we want to delete all the offers and exit here because we don't want the bot to run if monitoring isn't working
+					// if monitoring is desired but not working properly, we want the bot to be shut down and guarantee that there
+					// aren't outstanding offers.
+					deleteAllOffersAndExit(botConfig, client, sdex)
+				}
+			}()
+		}
+		// --- end initialization of services ---
+
 		log.Println("Starting the trader bot...")
 		for {
 			bot.Start()
@@ -144,6 +162,26 @@ func init() {
 	}
 }
 
+func startMonitoringServer(botConfig trader.BotConfig) error {
+	healthMetrics, e := monitoring.MakeMetricsRecorder(map[string]interface{}{"success": true})
+	if e != nil {
+		return fmt.Errorf("unable to make metrics recorder for the health endpoint: %s", e)
+	}
+
+	healthEndpoint, e := monitoring.MakeMetricsEndpoint("/health", healthMetrics, networking.NoAuth)
+	if e != nil {
+		return fmt.Errorf("unable to make /health endpoint: %s", e)
+	}
+
+	server, e := networking.MakeServer([]networking.Endpoint{healthEndpoint})
+	if e != nil {
+		return fmt.Errorf("unable to initialize the metrics server: %s", e)
+	}
+
+	log.Printf("Starting monitoring server on port %d\n", botConfig.MonitoringPort)
+	return server.StartServer(botConfig.MonitoringPort, "", "")
+}
+
 func validateTrustlines(client *horizon.Client, botConfig *trader.BotConfig) {
 	account, e := client.LoadAccount(botConfig.TradingAccount())
 	if e != nil {

examples/configs/trader/sample_trader.cfg

@@ -19,3 +19,6 @@ ISSUER_B="GBMMZMK2DC4FFP4CAI6KCVNCQ7WLO5A7DQU7EC7WGHRDQBZB763X4OQI"
 TICK_INTERVAL_SECONDS=300
 # the url for your horizon instance. If this url contains the string "test" then the bot assumes it is using the test network.
 HORIZON_URL="https://horizon-testnet.stellar.org"
+
+# the port that the monitoring server should run on. Uncomment the following line to add monitoring server.
+# MONITORING_PORT=8081

glide.lock

@@ -17,3 +17,5 @@ import:
   - support/errors
 - package: github.com/PagerDuty/go-pagerduty
   version: 635c5ce271490fba94880e62cde4eea3c1c184b9
+- package: github.com/go-chi/chi
+  version: 0ebf7795c516423a110473652e9ba3a59a504863

glide.yaml

@@ -22,4 +22,4 @@ func MakeAlert(alertType string, apiKey string) (api.Alert, error) {
 	default:
 		return &noopAlert{}, nil
 	}
-}
+}

support/monitoring/factory.go

@@ -0,0 +1,8 @@
+package monitoring
+
+// Metrics is an interface that allows a client to pass in key value pairs (keys must be strings)
+// and it can dump the metrics as JSON.
+type Metrics interface {
+	UpdateMetrics(metrics map[string]interface{})
+	MarshalJSON() ([]byte, error)
+}

support/monitoring/metrics.go

@@ -0,0 +1,60 @@
+package monitoring
+
+import (
+	"fmt"
+	"log"
+	"net/http"
+	"strings"
+
+	"github.com/lightyeario/kelp/support/networking"
+)
+
+// metricsEndpoint represents a monitoring API endpoint that always responds with a JSON
+// encoding of the provided metrics. The auth level for the endpoint can be NoAuth (public access)
+// or GoogleAuth which uses a Google account for authorization.
+type metricsEndpoint struct {
+	path      string
+	metrics   Metrics
+	authLevel networking.AuthLevel
+}
+
+// MakeMetricsEndpoint creates an Endpoint for the monitoring server with the desired auth level.
+// The endpoint's response is always a JSON dump of the provided metrics.
+func MakeMetricsEndpoint(path string, metrics Metrics, authLevel networking.AuthLevel) (networking.Endpoint, error) {
+	if !strings.HasPrefix(path, "/") {
+		return nil, fmt.Errorf("endpoint path must begin with /")
+	}
+	s := &metricsEndpoint{
+		path:      path,
+		metrics:   metrics,
+		authLevel: authLevel,
+	}
+	return s, nil
+}
+
+func (m *metricsEndpoint) GetAuthLevel() networking.AuthLevel {
+	return m.authLevel
+}
+
+func (m *metricsEndpoint) GetPath() string {
+	return m.path
+}
+
+// GetHandlerFunc returns a HandlerFunc that writes the JSON representation of the metrics
+// that's passed into the endpoint.
+func (m *metricsEndpoint) GetHandlerFunc() http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		json, e := m.metrics.MarshalJSON()
+		if e != nil {
+			log.Printf("error marshalling metrics json: %s\n", e)
+			http.Error(w, e.Error(), 500)
+			return
+		}
+		w.WriteHeader(200)
+		w.Header().Set("Content-Type", "application/json")
+		_, e = w.Write(json)
+		if e != nil {
+			log.Printf("error writing to the response writer: %s\n", e)
+		}
+	}
+}

support/monitoring/metricsEndpoint.go

@@ -0,0 +1,39 @@
+package monitoring
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/lightyeario/kelp/support/networking"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMetricsEndpoint_NoAuthEndpoint(t *testing.T) {
+	testMetrics, e := MakeMetricsRecorder(map[string]interface{}{"this is a test message": true})
+	if !assert.Nil(t, e) {
+		return
+	}
+	testEndpoint, e := MakeMetricsEndpoint("/test", testMetrics, networking.NoAuth)
+	if !assert.Nil(t, e) {
+		return
+	}
+
+	req, e := http.NewRequest("GET", "/test", nil)
+	if !assert.Nil(t, e) {
+		return
+	}
+	w := httptest.NewRecorder()
+	testEndpoint.GetHandlerFunc().ServeHTTP(w, req)
+	assert.Equal(t, 200, w.Code)
+	assert.Equal(t, "{\"this is a test message\":true}", w.Body.String())
+
+	// Mutate the metrics and test if the server response changes
+	testMetrics.UpdateMetrics(map[string]interface{}{"this is a test message": false})
+
+	w = httptest.NewRecorder()
+	req = httptest.NewRequest("GET", "/test", nil)
+	testEndpoint.GetHandlerFunc().ServeHTTP(w, req)
+	assert.Equal(t, 200, w.Code)
+	assert.Equal(t, "{\"this is a test message\":false}", w.Body.String())
+}

support/monitoring/metricsEndpoint_test.go

@@ -0,0 +1,36 @@
+package monitoring
+
+import "encoding/json"
+
+// MetricsRecorder uses a map to store metrics and implements the api.Metrics interface.
+type metricsRecorder struct {
+	records map[string]interface{}
+}
+
+var _ Metrics = &metricsRecorder{}
+
+// MakeMetricsRecorder makes a metrics recorder with the records map as the underlying map. If records
+// is nil, then an empty map will be initialized for you.
+func MakeMetricsRecorder(records map[string]interface{}) (Metrics, error) {
+	if records == nil {
+		return &metricsRecorder{
+			records: map[string]interface{}{},
+		}, nil
+	}
+	return &metricsRecorder{
+		records: records,
+	}, nil
+}
+
+// UpdateMetrics updates (or adds if non-existent) metrics in the records for all key-value
+// pairs in the provided map of metrics.
+func (m *metricsRecorder) UpdateMetrics(metrics map[string]interface{}) {
+	for k, v := range metrics {
+		m.records[k] = v
+	}
+}
+
+// MarshalJSON gives the JSON representation of the records.
+func (m *metricsRecorder) MarshalJSON() ([]byte, error) {
+	return json.Marshal(m.records)
+}

support/monitoring/metricsRecorder.go

@@ -0,0 +1,45 @@
+package monitoring
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMetricsRecorder_UpdateMetrics(t *testing.T) {
+	m := &metricsRecorder{
+		records: map[string]interface{}{},
+	}
+	m.UpdateMetrics(map[string]interface{}{
+		"max_qty":      100,
+		"volume":       200000,
+		"kelp_version": "1.1",
+	})
+	assert.Equal(t, 100, m.records["max_qty"])
+	assert.Equal(t, 200000, m.records["volume"])
+	assert.Equal(t, "1.1", m.records["kelp_version"])
+	assert.Equal(t, nil, m.records["nonexistent"])
+	m.UpdateMetrics(map[string]interface{}{
+		"max_qty": 200,
+	})
+	assert.Equal(t, 200, m.records["max_qty"])
+}
+
+func TestMetricsRecorder_MarshalJSON(t *testing.T) {
+	m := &metricsRecorder{
+		records: map[string]interface{}{},
+	}
+	m.UpdateMetrics(map[string]interface{}{
+		"statuses": map[string]string{
+			"a": "ok",
+			"b": "error",
+		},
+		"trade_ids": []int64{1, 2, 3, 4, 5},
+		"version":   "10.0.1",
+	})
+	json, e := m.MarshalJSON()
+	if !assert.Nil(t, e) {
+		return
+	}
+	assert.Equal(t, `{"statuses":{"a":"ok","b":"error"},"trade_ids":[1,2,3,4,5],"version":"10.0.1"}`, string(json))
+}