From 2b6854e4735e03838ea70645846c7e7b9ffb5d3e Mon Sep 17 00:00:00 2001 From: Joshua Gilman Date: Wed, 7 Oct 2026 15:20:15 -0700 Subject: [PATCH 1/4] node: report Prometheus remote write rejections as errors A non-2xx response from the remote write endpoint was logged and then reported as success, so an invalid token or rejected batch looked healthy to the caller. Return an error carrying the status and the start of the response body for any non-2xx response. --- .../telemetry/prom_remote_write/prom_test.go | 35 +++++++++++++++++++ .../pkg/telemetry/prom_remote_write/scrape.go | 13 +++++-- 2 files changed, 45 insertions(+), 3 deletions(-) diff --git a/node/pkg/telemetry/prom_remote_write/prom_test.go b/node/pkg/telemetry/prom_remote_write/prom_test.go index 838ebd159c3..644a7f05dbc 100644 --- a/node/pkg/telemetry/prom_remote_write/prom_test.go +++ b/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) + }) + } +} diff --git a/node/pkg/telemetry/prom_remote_write/scrape.go b/node/pkg/telemetry/prom_remote_write/scrape.go index ad2e3fd309b..aa5415a96db 100644 --- a/node/pkg/telemetry/prom_remote_write/scrape.go +++ b/node/pkg/telemetry/prom_remote_write/scrape.go @@ -3,8 +3,11 @@ 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" @@ -12,6 +15,9 @@ import ( "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 } From a4faf8ebebd9e5dc4fe7c2719e2fede376b1d6e7 Mon Sep 17 00:00:00 2001 From: Joshua Gilman Date: Wed, 7 Oct 2026 15:25:22 -0700 Subject: [PATCH 2/4] node: add --promRemoteInterval for Prometheus remote write The push interval was fixed at 15 seconds. Hosted receivers bill by samples per minute as well as active series, so a fixed 15 s interval costs four times what a 1 minute interval does for the same series. Default stays 15 s; non-positive values are rejected at startup. Also document --promRemoteURL in operations.md, and correct the statement that Loki telemetry is optional on mainnet. --- docs/guardian.md | 2 +- docs/operations.md | 11 ++++++++--- node/cmd/guardiand/node.go | 13 ++++++++++--- 3 files changed, 19 insertions(+), 7 deletions(-) diff --git a/docs/guardian.md b/docs/guardian.md index 8a38ac5cb61..2342a60943d 100644 --- a/docs/guardian.md +++ b/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 diff --git a/docs/operations.md b/docs/operations.md index d9a1418e270..a8ab64c8b91 100644 --- a/docs/operations.md +++ b/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 diff --git a/node/cmd/guardiand/node.go b/node/cmd/guardiand/node.go index e67f0cec634..6e91c897c85 100644 --- a/node/cmd/guardiand/node.go +++ b/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 { From 0864515b89c6d16282beca76f011b5912ef41c1f Mon Sep 17 00:00:00 2001 From: Joshua Gilman Date: Thu, 8 Oct 2026 09:14:45 -0700 Subject: [PATCH 3/4] node: add common.SafeReadN for a caller-chosen read limit SafeRead caps reads at MaxSafeInputSize (128MB). The remote write error path wrapped it in an io.LimitReader to keep the logged body small. SafeReadN takes the limit as an argument (1..MaxSafeInputSize) and SafeRead now delegates to it, so existing callers are unchanged. A rejected remote write with a body over 512 bytes now reports the status without the body instead of the first 512 bytes. --- node/pkg/common/chainlock.go | 27 ++++++++++++---- node/pkg/common/chainlock_test.go | 32 +++++++++++++++++++ .../telemetry/prom_remote_write/prom_test.go | 2 ++ .../pkg/telemetry/prom_remote_write/scrape.go | 6 ++-- 4 files changed, 58 insertions(+), 9 deletions(-) diff --git a/node/pkg/common/chainlock.go b/node/pkg/common/chainlock.go index 86d42c51c0d..07903c2483e 100644 --- a/node/pkg/common/chainlock.go +++ b/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 } diff --git a/node/pkg/common/chainlock_test.go b/node/pkg/common/chainlock_test.go index 56d636c81f0..f552a1c9b50 100644 --- a/node/pkg/common/chainlock_test.go +++ b/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 ( diff --git a/node/pkg/telemetry/prom_remote_write/prom_test.go b/node/pkg/telemetry/prom_remote_write/prom_test.go index 644a7f05dbc..04a442754cf 100644 --- a/node/pkg/telemetry/prom_remote_write/prom_test.go +++ b/node/pkg/telemetry/prom_remote_write/prom_test.go @@ -5,6 +5,7 @@ import ( "context" "net/http" "net/http/httptest" + "strings" "testing" prometheusv1 "github.com/certusone/wormhole/node/pkg/proto/prometheus/v1" @@ -221,6 +222,7 @@ func TestScrapeAndSendLocalMetricsReportsReceiverStatus(t *testing.T) { {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) { diff --git a/node/pkg/telemetry/prom_remote_write/scrape.go b/node/pkg/telemetry/prom_remote_write/scrape.go index aa5415a96db..d835c782c76 100644 --- a/node/pkg/telemetry/prom_remote_write/scrape.go +++ b/node/pkg/telemetry/prom_remote_write/scrape.go @@ -4,7 +4,6 @@ import ( "bytes" "context" "fmt" - "io" "net/http" "github.com/certusone/wormhole/node/pkg/common" @@ -85,7 +84,10 @@ func ScrapeAndSendLocalMetrics(ctx context.Context, info PromTelemetryInfo, logg logger.Debug("Grafana result", zap.Int("status code", res.StatusCode)) 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)) + 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 From 7620bd5dd90ecf392341d78accc1ebfbf442747d Mon Sep 17 00:00:00 2001 From: Joshua Gilman Date: Tue, 13 Oct 2026 08:03:46 -0700 Subject: [PATCH 4/4] node: require --promRemoteInterval of at least 15s Each push gathers and sends the whole metrics registry, so a very short interval loads both the node and the receiver. The flag exists to push less often; make the 15s default also the minimum. --- docs/operations.md | 2 +- node/cmd/guardiand/node.go | 10 +++++++--- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/docs/operations.md b/docs/operations.md index a8ab64c8b91..c3932c9d62e 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -715,7 +715,7 @@ Add the following flags: ``` Metrics are pushed every 15 seconds by default. Use `--promRemoteInterval` (for example `--promRemoteInterval=1m`) to -change the interval. +push less often; the minimum is 15 seconds. New guardians should talk to the Wormhole Foundation to get these URLs. diff --git a/node/cmd/guardiand/node.go b/node/cmd/guardiand/node.go index 6e91c897c85..ba7bf824760 100644 --- a/node/cmd/guardiand/node.go +++ b/node/cmd/guardiand/node.go @@ -563,7 +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)") + 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") @@ -600,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 @@ -1287,8 +1291,8 @@ func runNode(cmd *cobra.Command, args []string) { } usingPromRemoteWrite := *promRemoteURL != "" if usingPromRemoteWrite { - if *promRemoteInterval <= 0 { - logger.Fatal("--promRemoteInterval must be positive", zap.Duration("promRemoteInterval", *promRemoteInterval)) + if *promRemoteInterval < minPromRemoteInterval { + logger.Fatal("--promRemoteInterval must be at least 15s", zap.Duration("promRemoteInterval", *promRemoteInterval)) } var info promremotew.PromTelemetryInfo