feat(node): add --promRemoteInterval and report remote write rejections

docs/guardian.md

@@ -119,7 +119,7 @@ also publishes to Grafana in a non-blocking manner.
 
 The guardian can also be configured to publish metrics to Grafana. To enable this, set the `promRemoteURL`
 config parameter to point at a Grafana endpoint. In this mode, metrics will be posted to Grafana every
-fifteen seconds, also in a non-blocking manner.
+fifteen seconds by default (configurable with `promRemoteInterval`), also in a non-blocking manner.
 
 ## Observation Lifecycle
 

docs/operations.md

@@ -705,14 +705,19 @@ can be operated behind a load balancer.
 
 ## Enabling Telemetry
 
-Optionally, the guardian can send telemetry to [Grafana Cloud Logs](https://grafana.com/products/cloud/logs/) aka "loki".
-To enable this functionality, add the following flag:
+On mainnet and testnet, the guardian sends its logs to [Grafana Cloud Logs](https://grafana.com/products/cloud/logs/) aka "loki"
+unless telemetry is disabled with `--disableTelemetry`. It can also push its Prometheus metrics to Grafana Cloud.
+Add the following flags:
 
 ```bash
 --telemetryLokiURL=$PER_GUARDIAN_LOKI_URL_WITH_TOKEN
+--promRemoteURL=$PER_GUARDIAN_PROMETHEUS_URL_WITH_TOKEN
 ```
 
-New guardians should talk to the Wormhole Foundation to get a Loki url.
+Metrics are pushed every 15 seconds by default. Use `--promRemoteInterval` (for example `--promRemoteInterval=1m`) to
+change the interval.
+
+New guardians should talk to the Wormhole Foundation to get these URLs.
 
 ### Binding to privileged ports
 

node/cmd/guardiand/node.go

@@ -288,8 +288,9 @@ var (
 	// Loki cloud logging parameters
 	telemetryLokiURL *string
 
-	// Prometheus remote write URL
-	promRemoteURL *string
+	// Prometheus remote write URL and push interval
+	promRemoteURL      *string
+	promRemoteInterval *time.Duration
 
 	chainGovernorEnabled      *bool
 	governorFlowCancelEnabled *bool
@@ -562,6 +563,7 @@ func init() {
 	telemetryLokiURL = NodeCmd.Flags().String("telemetryLokiURL", "", "Loki cloud logging URL")
 
 	promRemoteURL = NodeCmd.Flags().String("promRemoteURL", "", "Prometheus remote write URL (Grafana)")
+	promRemoteInterval = NodeCmd.Flags().Duration("promRemoteInterval", 15*time.Second, "Interval between Prometheus remote write pushes (e.g. 15s, 1m)")
 
 	chainGovernorEnabled = NodeCmd.Flags().Bool("chainGovernorEnabled", false, "Run the chain governor")
 	governorFlowCancelEnabled = NodeCmd.Flags().Bool("governorFlowCancelEnabled", false, "Enable flow cancel on the governor")
@@ -1285,6 +1287,10 @@ func runNode(cmd *cobra.Command, args []string) {
 	}
 	usingPromRemoteWrite := *promRemoteURL != ""
 	if usingPromRemoteWrite {
+		if *promRemoteInterval <= 0 {
+			logger.Fatal("--promRemoteInterval must be positive", zap.Duration("promRemoteInterval", *promRemoteInterval))
+		}
+
 		var info promremotew.PromTelemetryInfo
 		info.PromRemoteURL = *promRemoteURL
 		info.Labels = map[string]string{
@@ -1296,9 +1302,10 @@ func runNode(cmd *cobra.Command, args []string) {
 		}
 
 		promLogger := logger.With(zap.String("component", "prometheus_scraper"))
+		promLogger.Info("Pushing metrics with Prometheus remote write", zap.Duration("interval", *promRemoteInterval))
 		errC := make(chan error)
 		common.StartRunnable(rootCtx, errC, false, "prometheus_scraper", func(ctx context.Context) error {
-			t := time.NewTicker(15 * time.Second)
+			t := time.NewTicker(*promRemoteInterval)
 
 			for {
 				select {

node/pkg/telemetry/prom_remote_write/prom_test.go

@@ -2,9 +2,13 @@ package promremotew
 
 import (
 	"bytes"
+	"context"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 
 	prometheusv1 "github.com/certusone/wormhole/node/pkg/proto/prometheus/v1"
+	"go.uber.org/zap"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/stretchr/testify/require"
@@ -205,3 +209,34 @@ func TestMarshalUnmarshal(t *testing.T) {
 	err = proto.Unmarshal(wrBytes, &newWr)
 	require.NoError(t, err)
 }
+
+func TestScrapeAndSendLocalMetricsReportsReceiverStatus(t *testing.T) {
+	tests := []struct {
+		name    string
+		status  int
+		body    string
+		wantErr string
+	}{
+		{name: "no content", status: http.StatusNoContent},
+		{name: "ok", status: http.StatusOK},
+		{name: "unauthorized", status: http.StatusUnauthorized, body: "authentication error: invalid token\n", wantErr: "remote write returned status 401: authentication error: invalid token"},
+		{name: "server error", status: http.StatusInternalServerError, wantErr: "remote write returned status 500"},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+				w.WriteHeader(tc.status)
+				_, _ = w.Write([]byte(tc.body))
+			}))
+			defer srv.Close()
+
+			info := PromTelemetryInfo{PromRemoteURL: srv.URL, Labels: map[string]string{"product": "wormhole"}}
+			err := ScrapeAndSendLocalMetrics(context.Background(), info, zap.NewNop())
+			if tc.wantErr == "" {
+				require.NoError(t, err)
+				return
+			}
+			require.ErrorContains(t, err, tc.wantErr)
+		})
+	}
+}

node/pkg/telemetry/prom_remote_write/scrape.go

@@ -3,15 +3,21 @@ package promremotew
 import (
 	"bytes"
 	"context"
+	"fmt"
+	"io"
 	"net/http"
 
+	"github.com/certusone/wormhole/node/pkg/common"
 	"github.com/golang/snappy"
 	"github.com/prometheus/client_golang/prometheus"
 	dto "github.com/prometheus/client_model/go"
 	"go.uber.org/zap"
 	"google.golang.org/protobuf/proto"
 )
 
+// maxErrorBodyBytes caps how much of a rejected remote write response is included in the error.
+const maxErrorBodyBytes = 512
+
 type PromTelemetryInfo struct {
 	PromRemoteURL string
 	Labels        map[string]string
@@ -77,9 +83,10 @@ func ScrapeAndSendLocalMetrics(ctx context.Context, info PromTelemetryInfo, logg
 	defer res.Body.Close()
 
 	logger.Debug("Grafana result", zap.Int("status code", res.StatusCode))
-	if res.StatusCode != 200 && res.StatusCode != 204 {
-		logger.Error("Grafana returned a status code other than 200 or 204", zap.Int("status code", res.StatusCode))
-		return err
+	if res.StatusCode < 200 || res.StatusCode > 299 {
+		// The response body carries the receiver's reason (e.g. an authentication error).
+		body, _ := common.SafeRead(io.LimitReader(res.Body, maxErrorBodyBytes))
+		return fmt.Errorf("remote write returned status %d: %s", res.StatusCode, bytes.TrimSpace(body))
 	}
 	return nil
 }