fix(grpc-proxy): classify NATS connection failures

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

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

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",
     ],

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 {

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

src/invocation-plane-services/grpc-proxy/proxy/metrics/metrics.go

@@ -6,7 +6,7 @@ Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at
 
-    http://www.apache.org/licenses/LICENSE-2.0
+	http://www.apache.org/licenses/LICENSE-2.0
 
 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
@@ -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,
@@ -301,15 +315,15 @@ var WorkerConnectionCloseReasons = []string{
 // Outcomes of a worker CONNECT to /v1/proxy. Every terminal path in
 // HijackHandler maps to exactly one of these.
 const (
-	ConnectAccepted            = "accepted"
-	ConnectNotHijackable       = "rejected_not_hijackable"    // 500
-	ConnectMissingAuth         = "rejected_missing_auth"      // 401
-	ConnectMissingRequestID    = "rejected_missing_requestid" // 400
-	ConnectInvalidRequestID    = "rejected_invalid_requestid" // 400
-	ConnectTokenExpired        = "rejected_token_expired"     // 403, token was issued but has aged out
-	ConnectTokenUnknown        = "rejected_token_unknown"     // 403, token was never issued by this pod
-	ConnectRequestIDMismatch   = "rejected_requestid_mismatch"// 403, token valid but bound to another request
-	ConnectHijackFailed        = "rejected_hijack_failed"     // 500
+	ConnectAccepted          = "accepted"
+	ConnectNotHijackable     = "rejected_not_hijackable"     // 500
+	ConnectMissingAuth       = "rejected_missing_auth"       // 401
+	ConnectMissingRequestID  = "rejected_missing_requestid"  // 400
+	ConnectInvalidRequestID  = "rejected_invalid_requestid"  // 400
+	ConnectTokenExpired      = "rejected_token_expired"      // 403, token was issued but has aged out
+	ConnectTokenUnknown      = "rejected_token_unknown"      // 403, token was never issued by this pod
+	ConnectRequestIDMismatch = "rejected_requestid_mismatch" // 403, token valid but bound to another request
+	ConnectHijackFailed      = "rejected_hijack_failed"      // 500
 )
 
 var ConnectResults = []string{
@@ -387,6 +401,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())
@@ -406,6 +440,9 @@ func init() {
 	for _, code := range CloseCodes {
 		WorkerConnectionCloseCodeTotal.WithLabelValues(code)
 	}
+	for _, reason := range NatsErrorReasons {
+		NatsFailureCounter.WithLabelValues(reason)
+	}
 }
 
 var nc atomic.Pointer[nats.Conn]