From c5a50d07486c28f51440661ba9a290aa8e90aa50 Mon Sep 17 00:00:00 2001 From: Bora Oztekin Date: Wed, 19 Aug 2026 21:05:22 +0000 Subject: [PATCH] fix(grpc-proxy): classify NATS connection failures Add disconnect and classified failure metrics while preserving the existing unlabeled NATS error counter contract. Tests use github.com/kylelemons/godebug v1.1.0 transitively through Prometheus testutil. The dependency is Apache-2.0 and is already represented in NOTICE. Signed-off-by: Bora Oztekin --- docs/user/metrics/grpc-proxy/metrics.md | 25 +++++- .../grpc-proxy/go.mod | 1 + .../grpc-proxy/proxy/invocation/BUILD.bazel | 3 + .../grpc-proxy/proxy/invocation/nats.go | 84 ++++++++++++++++++- .../grpc-proxy/proxy/invocation/nats_test.go | 47 +++++++++++ .../grpc-proxy/proxy/metrics/metrics.go | 37 ++++++++ 6 files changed, 193 insertions(+), 4 deletions(-) diff --git a/docs/user/metrics/grpc-proxy/metrics.md b/docs/user/metrics/grpc-proxy/metrics.md index 2af89249a..adaef77c7 100644 --- a/docs/user/metrics/grpc-proxy/metrics.md +++ b/docs/user/metrics/grpc-proxy/metrics.md @@ -13,6 +13,8 @@ The metric families below cover client connection health, the NATS pipe to worke | nvcf_grpc_proxy_service_nats_out_bytes | Gauge | grpc:10083/metrics | Bytes sent to the NATS connection | bytes | | namespace="nvcf" | | nvcf_grpc_proxy_service_nats_out_msgs | Gauge | grpc:10083/metrics | Messages sent to the NATS connection | | | namespace="nvcf" | | nvcf_grpc_proxy_service_nats_error_total | Counter | grpc:10083/metrics | Errors observed on the NATS connection | | | namespace="nvcf" | +| nvcf_grpc_proxy_service_nats_failure_total | Counter | grpc:10083/metrics | NATS failures classified by cause | | reason | namespace="nvcf" | +| nvcf_grpc_proxy_service_nats_disconnect_total | Counter | grpc:10083/metrics | NATS disconnect events | | | namespace="nvcf" | | nvcf_grpc_proxy_service_nats_reconnect_total | Counter | grpc:10083/metrics | NATS reconnect attempts | | | namespace="nvcf" | | nvcf_grpc_proxy_service_nats_reconnects | Gauge | grpc:10083/metrics | Current reconnect attempt count | | | namespace="nvcf" | | nvcf_grpc_proxy_service_nats_lame_duck_total | Counter | grpc:10083/metrics | NATS lame-duck messages observed | | | namespace="nvcf" | @@ -24,11 +26,30 @@ The metric families below cover client connection health, the NATS pipe to worke ## Notes -- `nvcf_grpc_proxy_service_session_init_seconds_bucket` is the SLI for **gRPC** inference function health. HTTP inference functions invoked through the regular HTTP invocation gateway bypass the gRPC proxy entirely and do not register session-init samples here. -- The `nvcf_grpc_proxy_service_nats_*` family is a useful proxy signal for "is the gRPC proxy → NATS pipe healthy" (in/out bytes and message deltas) and "are NATS upstreams stable" (reconnect and error counters). +- `nvcf_grpc_proxy_service_session_init_seconds_bucket` is the SLI for gRPC inference function health. HTTP inference functions invoked through the regular HTTP invocation gateway bypass the gRPC proxy entirely and do not register session-init samples here. +- The `nvcf_grpc_proxy_service_nats_*` family is a useful proxy signal for "is the gRPC proxy -> NATS pipe healthy" (in/out bytes and message deltas) and "are NATS upstreams stable" (reconnect and error counters). - Per-RPC outcomes (success vs. error per call) are covered by the OpenTelemetry `rpc_client_*` family; aggregate proxy-side errors are covered by `nvcf_grpc_proxy_service_nats_error_total`. +- The NATS failure `reason` values are `certificate_expired`, `tls_verification`, `tls`, `authentication`, `timeout`, `connection`, and `other`. - Standard Go runtime metrics (`go_*`) and process metrics (`process_*`) are also exposed on the same endpoint and follow upstream conventions. +## Alert queries + +Alert on disconnect churn independently from successful reconnects: + +```promql +sum(increase(nvcf_grpc_proxy_service_nats_disconnect_total[5m])) > 0 +``` + +Group NATS errors by their bounded reason label: + +```promql +sum by (reason) (increase(nvcf_grpc_proxy_service_nats_failure_total[5m])) > 0 +``` + +Use `reason="certificate_expired"` for certificate expiry alerts. Use +`reason=~"tls_verification|tls"` for other TLS failures and +`reason="authentication"` for NATS credential failures. + ## Reproducing locally ```bash diff --git a/src/invocation-plane-services/grpc-proxy/go.mod b/src/invocation-plane-services/grpc-proxy/go.mod index 2163b7c0f..3942bfe7c 100644 --- a/src/invocation-plane-services/grpc-proxy/go.mod +++ b/src/invocation-plane-services/grpc-proxy/go.mod @@ -102,6 +102,7 @@ require ( github.com/hashicorp/golang-lru v1.0.2 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/klauspost/compress v1.18.0 // indirect + github.com/kylelemons/godebug v1.1.0 // indirect github.com/lufia/plan9stats v0.0.0-20240226150601-1dcf7310316a // indirect github.com/magiconair/properties v1.8.10 // indirect github.com/mattn/go-colorable v0.1.14 // indirect diff --git a/src/invocation-plane-services/grpc-proxy/proxy/invocation/BUILD.bazel b/src/invocation-plane-services/grpc-proxy/proxy/invocation/BUILD.bazel index 8d5ef783b..11c79d628 100644 --- a/src/invocation-plane-services/grpc-proxy/proxy/invocation/BUILD.bazel +++ b/src/invocation-plane-services/grpc-proxy/proxy/invocation/BUILD.bazel @@ -60,7 +60,10 @@ go_test( srcs = ["nats_test.go"], embed = [":invocation"], deps = [ + "//src/invocation-plane-services/grpc-proxy/proxy/metrics", + "@com_github_nats_io_nats_go//:nats_go", "@com_github_nats_io_nkeys//:nkeys", + "@com_github_prometheus_client_golang//prometheus/testutil", "@com_github_stretchr_testify//assert", "@com_github_stretchr_testify//require", ], diff --git a/src/invocation-plane-services/grpc-proxy/proxy/invocation/nats.go b/src/invocation-plane-services/grpc-proxy/proxy/invocation/nats.go index 3baf1ec79..7c29f968f 100644 --- a/src/invocation-plane-services/grpc-proxy/proxy/invocation/nats.go +++ b/src/invocation-plane-services/grpc-proxy/proxy/invocation/nats.go @@ -18,9 +18,14 @@ package invocation import ( "context" + "crypto/tls" + "crypto/x509" "encoding/base64" "encoding/json" + "errors" + "io" "math/rand" + "net" "strings" "time" @@ -64,8 +69,11 @@ func NewNatsConnection(natsFqdn, nKeySeed, serviceName, ssaFqdn, secretsPath str metrics.NatsReconnectCounter.Inc() }), nats.Name(serviceName), nats.ErrorHandler(func(nc *nats.Conn, sub *nats.Subscription, err error) { - zap.L().Warn("nats connection error", zap.Error(err)) - metrics.NatsErrorCounter.Inc() + recordNatsAsyncError(err) + }), nats.ReconnectErrHandler(func(conn *nats.Conn, err error) { + recordNatsFailure("nats reconnect failed", err) + }), nats.DisconnectErrHandler(func(conn *nats.Conn, err error) { + recordNatsDisconnect(err) }), nats.ConnectHandler(func(conn *nats.Conn) { zap.L().Info("connected to nats", zap.String("server", conn.ConnectedServerName()), zap.String("cluster", conn.ConnectedClusterName())) })) @@ -76,6 +84,78 @@ func NewNatsConnection(natsFqdn, nKeySeed, serviceName, ssaFqdn, secretsPath str return nc, nil } +func recordNatsDisconnect(err error) { + if err == nil { + zap.L().Info("disconnected from nats") + } else { + zap.L().Warn("disconnected from nats", zap.Error(err)) + } + metrics.NatsDisconnectCounter.Inc() +} + +func recordNatsAsyncError(err error) { + metrics.NatsErrorCounter.Inc() + recordNatsFailure("nats connection error", err) +} + +func recordNatsFailure(message string, err error) { + reason := natsErrorReason(err) + zap.L().Warn(message, zap.String("reason", reason), zap.Error(err)) + metrics.NatsFailureCounter.WithLabelValues(reason).Inc() +} + +func natsErrorReason(err error) string { + var certificateInvalid x509.CertificateInvalidError + if errors.As(err, &certificateInvalid) { + if certificateInvalid.Reason == x509.Expired { + return metrics.NatsErrorReasonCertificateExpired + } + return metrics.NatsErrorReasonTLSVerification + } + + var unknownAuthority x509.UnknownAuthorityError + var hostnameError x509.HostnameError + var certificateVerification *tls.CertificateVerificationError + if errors.As(err, &unknownAuthority) || errors.As(err, &hostnameError) || errors.As(err, &certificateVerification) { + return metrics.NatsErrorReasonTLSVerification + } + + errorText := strings.ToLower(errString(err)) + if strings.Contains(errorText, "certificate has expired") || strings.Contains(errorText, "expired certificate") { + return metrics.NatsErrorReasonCertificateExpired + } + if errors.Is(err, nats.ErrSecureConnRequired) || errors.Is(err, nats.ErrSecureConnWanted) || + strings.Contains(errorText, "tls") || strings.Contains(errorText, "x509") { + return metrics.NatsErrorReasonTLS + } + + if errors.Is(err, nats.ErrAuthorization) || errors.Is(err, nats.ErrAuthExpired) || + errors.Is(err, nats.ErrAuthRevoked) || errors.Is(err, nats.ErrAccountAuthExpired) { + return metrics.NatsErrorReasonAuthentication + } + + var networkError net.Error + if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, nats.ErrTimeout) || + (errors.As(err, &networkError) && networkError.Timeout()) { + return metrics.NatsErrorReasonTimeout + } + + if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) || + errors.Is(err, nats.ErrConnectionClosed) || errors.Is(err, nats.ErrConnectionReconnecting) || + errors.Is(err, nats.ErrDisconnected) || errors.Is(err, nats.ErrNoServers) || + errors.Is(err, nats.ErrStaleConnection) || errors.As(err, &networkError) { + return metrics.NatsErrorReasonConnection + } + return metrics.NatsErrorReasonOther +} + +func errString(err error) string { + if err == nil { + return "" + } + return err.Error() +} + func newNkeyAuthOption(nKeySeed string) (nats.Option, error) { kp, err := nkeys.FromSeed([]byte(nKeySeed)) if err != nil { diff --git a/src/invocation-plane-services/grpc-proxy/proxy/invocation/nats_test.go b/src/invocation-plane-services/grpc-proxy/proxy/invocation/nats_test.go index 3cda054e1..fa7cc6e5b 100644 --- a/src/invocation-plane-services/grpc-proxy/proxy/invocation/nats_test.go +++ b/src/invocation-plane-services/grpc-proxy/proxy/invocation/nats_test.go @@ -17,17 +17,64 @@ limitations under the License. package invocation import ( + "context" + "crypto/x509" "encoding/base64" "encoding/json" + "errors" + "fmt" + "io" "os" "path/filepath" "testing" + "github.com/nats-io/nats.go" "github.com/nats-io/nkeys" + "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + + "nvcf-grpc-proxy/proxy/metrics" ) +func TestNatsErrorReason(t *testing.T) { + tests := []struct { + name string + err error + want string + }{ + {name: "expired certificate", err: x509.CertificateInvalidError{Reason: x509.Expired}, want: metrics.NatsErrorReasonCertificateExpired}, + {name: "verification failure", err: x509.UnknownAuthorityError{}, want: metrics.NatsErrorReasonTLSVerification}, + {name: "generic tls failure", err: errors.New("remote error: tls: bad certificate"), want: metrics.NatsErrorReasonTLS}, + {name: "authentication failure", err: nats.ErrAuthorization, want: metrics.NatsErrorReasonAuthentication}, + {name: "timeout", err: context.DeadlineExceeded, want: metrics.NatsErrorReasonTimeout}, + {name: "connection failure", err: io.EOF, want: metrics.NatsErrorReasonConnection}, + {name: "other failure", err: errors.New("unexpected error"), want: metrics.NatsErrorReasonOther}, + {name: "wrapped error", err: fmt.Errorf("reconnect: %w", nats.ErrAuthExpired), want: metrics.NatsErrorReasonAuthentication}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.want, natsErrorReason(test.err)) + }) + } +} + +func TestRecordNatsDisconnect(t *testing.T) { + before := testutil.ToFloat64(metrics.NatsDisconnectCounter) + recordNatsDisconnect(io.EOF) + assert.Equal(t, before+1, testutil.ToFloat64(metrics.NatsDisconnectCounter)) +} + +func TestRecordNatsAsyncError(t *testing.T) { + failureCounter := metrics.NatsFailureCounter.WithLabelValues(metrics.NatsErrorReasonTimeout) + errorBefore := testutil.ToFloat64(metrics.NatsErrorCounter) + failureBefore := testutil.ToFloat64(failureCounter) + recordNatsAsyncError(context.DeadlineExceeded) + assert.Equal(t, errorBefore+1, testutil.ToFloat64(metrics.NatsErrorCounter)) + assert.Equal(t, failureBefore+1, testutil.ToFloat64(failureCounter)) +} + func TestNewNkeyAuthOption(t *testing.T) { t.Run("valid nkey seed", func(t *testing.T) { // Generate a valid nkey seed for testing diff --git a/src/invocation-plane-services/grpc-proxy/proxy/metrics/metrics.go b/src/invocation-plane-services/grpc-proxy/proxy/metrics/metrics.go index 48381267c..f61b8692e 100644 --- a/src/invocation-plane-services/grpc-proxy/proxy/metrics/metrics.go +++ b/src/invocation-plane-services/grpc-proxy/proxy/metrics/metrics.go @@ -67,6 +67,20 @@ var ( Help: "total nats errors on a nats connection", }) + NatsFailureCounter = promauto.NewCounterVec( + prometheus.CounterOpts{ + Namespace: NatsNamespace, + Name: "failure_total", + Help: "total nats failures, by reason", + }, []string{"reason"}) + + NatsDisconnectCounter = promauto.NewCounter( + prometheus.CounterOpts{ + Namespace: NatsNamespace, + Name: "disconnect_total", + Help: "total nats disconnect events", + }) + NatsReconnectCounter = promauto.NewCounter( prometheus.CounterOpts{ Namespace: NatsNamespace, @@ -307,6 +321,26 @@ var ( }) ) +const ( + NatsErrorReasonCertificateExpired = "certificate_expired" + NatsErrorReasonTLSVerification = "tls_verification" + NatsErrorReasonTLS = "tls" + NatsErrorReasonAuthentication = "authentication" + NatsErrorReasonTimeout = "timeout" + NatsErrorReasonConnection = "connection" + NatsErrorReasonOther = "other" +) + +var NatsErrorReasons = []string{ + NatsErrorReasonCertificateExpired, + NatsErrorReasonTLSVerification, + NatsErrorReasonTLS, + NatsErrorReasonAuthentication, + NatsErrorReasonTimeout, + NatsErrorReasonConnection, + NatsErrorReasonOther, +} + func init() { // Set up OpenTelemetry metrics with Prometheus exporter exporter := lo.Must(otelprom.New()) @@ -323,6 +357,9 @@ func init() { for _, result := range ConnectResults { WorkerConnectTotal.WithLabelValues(result) } + for _, reason := range NatsErrorReasons { + NatsFailureCounter.WithLabelValues(reason) + } } var nc atomic.Pointer[nats.Conn]