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
+push less often; the minimum is 15 seconds.
+
+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", minPromRemoteInterval, "Interval between Prometheus remote write pushes (minimum and default 15s, e.g. 1m)")
 
 	chainGovernorEnabled = NodeCmd.Flags().Bool("chainGovernorEnabled", false, "Run the chain governor")
 	governorFlowCancelEnabled = NodeCmd.Flags().Bool("governorFlowCancelEnabled", false, "Enable flow cancel on the governor")
@@ -598,6 +600,10 @@ var (
 
 const envPrefix = "GUARDIAND"
 
+// minPromRemoteInterval is the default and the shortest allowed --promRemoteInterval. Each push
+// gathers and sends the whole metrics registry, so the interval may only be lengthened.
+const minPromRemoteInterval = 15 * time.Second
+
 // "Why would anyone do this?" are famous last words.
 //
 // We already forcibly override RPC URLs and keys in dev mode to prevent security
@@ -1285,6 +1291,10 @@ func runNode(cmd *cobra.Command, args []string) {
 	}
 	usingPromRemoteWrite := *promRemoteURL != ""
 	if usingPromRemoteWrite {
+		if *promRemoteInterval < minPromRemoteInterval {
+			logger.Fatal("--promRemoteInterval must be at least 15s", zap.Duration("promRemoteInterval", *promRemoteInterval))
+		}
+
 		var info promremotew.PromTelemetryInfo
 		info.PromRemoteURL = *promRemoteURL
 		info.Labels = map[string]string{
@@ -1296,9 +1306,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/common/chainlock.go

@@ -100,6 +100,8 @@ const MaxSafeInputSize = 128 * 1024 * 1024 // 128MB (arbitrary)
 
 var ErrInputTooLarge = errors.New("input data exceeds maximum allowed size")
 
+var ErrInvalidReadLimit = errors.New("read limit must be between 1 and MaxSafeInputSize")
+
 var (
 	ErrBinaryWrite         = errors.New("failed to write binary data")
 	ErrTxIDTooLong         = errors.New("field TxID too long")
@@ -762,25 +764,36 @@ func validBinaryBool(b byte) bool {
 // SafeRead reads from r with a size limit to prevent memory exhaustion attacks.
 // It returns an error if the input exceeds MaxSafeInputSize.
 func SafeRead(r io.Reader) ([]byte, error) {
-	// Create a LimitReader that allows reading up to MaxSafeInputSize + 1 bytes.
-	// The extra byte is specifically to detect if the input stream *exceeds* MaxSafeInputSize.
-	lr := io.LimitReader(r, MaxSafeInputSize+1)
+	return SafeReadN(r, MaxSafeInputSize)
+}
+
+// SafeReadN is SafeRead with a caller-chosen limit of n bytes, for callers that need a tighter
+// bound than MaxSafeInputSize. It returns ErrInputTooLarge if the input exceeds n bytes, and
+// ErrInvalidReadLimit if n is not between 1 and MaxSafeInputSize.
+func SafeReadN(r io.Reader, n int) ([]byte, error) {
+	if n < 1 || n > MaxSafeInputSize {
+		return nil, ErrInvalidReadLimit
+	}
+
+	// Create a LimitReader that allows reading up to n + 1 bytes.
+	// The extra byte is specifically to detect if the input stream *exceeds* n.
+	lr := io.LimitReader(r, int64(n)+1)
 
-	//nolint:forbidigo // SafeRead is intended as a convenient and safe wrapper for ReadAll.
+	//nolint:forbidigo // SafeReadN is intended as a convenient and safe wrapper for ReadAll.
 	b, err := io.ReadAll(lr)
 	if err != nil {
 		// Propagate any actual read errors from the underlying reader.
 		return nil, err
 	}
 
-	// If the length of the read bytes is greater than MaxSafeInputSize,
+	// If the length of the read bytes is greater than n,
 	// it means the original reader contained more data than allowed.
 	// In this case, we return an error instead of silently truncating.
-	if len(b) > MaxSafeInputSize {
+	if len(b) > n {
 		return nil, ErrInputTooLarge
 	}
 
-	// If err was nil and len(b) <= MaxSafeInputSize, it means we read all
+	// If err was nil and len(b) <= n, it means we read all
 	// available input (or up to the limit) without exceeding the maximum.
 	return b, nil
 }

node/pkg/common/chainlock_test.go

@@ -857,6 +857,38 @@ func TestSafeRead(t *testing.T) {
 	}
 }
 
+func TestSafeReadN(t *testing.T) {
+	const limit = 512
+	tests := []struct {
+		name    string
+		size    int
+		n       int
+		wantErr error
+	}{
+		{name: "empty input", size: 0, n: limit},
+		{name: "exactly at limit", size: limit, n: limit},
+		{name: "one byte over limit", size: limit + 1, n: limit, wantErr: ErrInputTooLarge},
+		{name: "limit of one", size: 1, n: 1},
+		{name: "zero limit", size: 1, n: 0, wantErr: ErrInvalidReadLimit},
+		{name: "negative limit", size: 1, n: -1, wantErr: ErrInvalidReadLimit},
+		{name: "limit above MaxSafeInputSize", size: 1, n: MaxSafeInputSize + 1, wantErr: ErrInvalidReadLimit},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			data := bytes.Repeat([]byte{0xab}, tt.size)
+
+			got, err := SafeReadN(bytes.NewReader(data), tt.n)
+			if tt.wantErr != nil {
+				require.ErrorIs(t, err, tt.wantErr)
+				require.Nil(t, got, "got should be nil when error occurs")
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, data, got)
+		})
+	}
+}
+
 func TestMessagePublication_IsWTT(t *testing.T) {
 	// Using real mainnet and testnet token bridge emitter addresses as hex strings
 	const (

node/pkg/telemetry/prom_remote_write/prom_test.go

@@ -2,9 +2,14 @@ package promremotew
 
 import (
 	"bytes"
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
 	"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 +210,35 @@ 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"},
+		{name: "oversized body", status: http.StatusBadGateway, body: strings.Repeat("x", maxErrorBodyBytes+1), wantErr: "remote write returned status 502 (response body not included: input data exceeds maximum allowed size)"},
+	}
+	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,20 @@ package promremotew
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"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 +82,13 @@ 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, err := common.SafeReadN(res.Body, maxErrorBodyBytes)
+		if err != nil {
+			return fmt.Errorf("remote write returned status %d (response body not included: %w)", res.StatusCode, err)
+		}
+		return fmt.Errorf("remote write returned status %d: %s", res.StatusCode, bytes.TrimSpace(body))
 	}
 	return nil
 }