Claude/fix issue 2893 x3rpd (#2895)

* feat: add prometheus monitoring wrapper

Reintroduces the Prometheus metrics wrapper previously available in the
plugins repository, updated for go-micro v5. Exposes request count and
latency histograms for handlers, subscribers, and outgoing client calls
via NewHandlerWrapper, NewSubscriberWrapper, NewCallWrapper and
NewClientWrapper, labelled with service/endpoint/status.

Options cover namespace, subsystem, const labels, histogram buckets and
a custom registerer; duplicate collectors (e.g. from multiple wrappers
sharing the same config) are reused transparently via a cached
metrics bundle.

Fixes #2893

* fix(registry/etcd): clear lease/register caches on KeepAlive channel closure

When the etcd client's long-lived KeepAlive channel closes (e.g. because
the lease expired on the server side during a network partition), the
previous cleanup only removed the channel bookkeeping. The stale entries
in `leases` and `register` caused the next registerNode() heartbeat to
hit the "unchanged hash" short-circuit and skip re-registration entirely,
so the service permanently disappeared from etcd.

Extract the cleanup into handleKeepAliveClosed and also drop the cached
lease id and hash so the next heartbeat performs a full Grant+Put and
the service recovers within one RegisterInterval.

Regression introduced by #2822; fix is symmetric with the existing
synchronous KeepAliveOnce recovery path that propagates
rpctypes.ErrLeaseNotFound.

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: asim

registry/etcd/etcd.go

@@ -14,9 +14,9 @@ import (
 	"time"
 
 	hash "github.com/mitchellh/hashstructure"
+	mtls "go-micro.dev/v5/internal/util/tls"
 	"go-micro.dev/v5/logger"
 	"go-micro.dev/v5/registry"
-	mtls "go-micro.dev/v5/internal/util/tls"
 	"go.etcd.io/etcd/api/v3/v3rpc/rpctypes"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.uber.org/zap"
@@ -468,13 +468,7 @@ func (e *etcdRegistry) startKeepAlive(key string, leaseID clientv3.LeaseID) erro
 			case ka, ok := <-ch:
 				if !ok {
 					log.Logf(logger.DebugLevel, "Keepalive channel closed for %s", key)
-					e.Lock()
-					// Only delete if still present (avoid race with stopKeepAlive)
-					if _, exists := e.keepaliveChs[key]; exists {
-						delete(e.keepaliveChs, key)
-						delete(e.keepaliveStop, key)
-					}
-					e.Unlock()
+					e.handleKeepAliveClosed(key)
 					return
 				}
 				if ka == nil {
@@ -489,6 +483,26 @@ func (e *etcdRegistry) startKeepAlive(key string, leaseID clientv3.LeaseID) erro
 	return nil
 }
 
+// handleKeepAliveClosed is invoked by the keepalive goroutine when the
+// etcd client closes the KeepAlive channel (for example because the lease
+// expired on the server side during a network partition). In addition to
+// removing the channel bookkeeping, it drops the cached lease and
+// registration hash so the next registerNode() performs a full
+// Grant+Put re-registration instead of being short-circuited by the
+// "unchanged" check at the top of registerNode.
+func (e *etcdRegistry) handleKeepAliveClosed(key string) {
+	e.Lock()
+	defer e.Unlock()
+
+	// Only delete if still present to avoid racing with stopKeepAlive.
+	if _, exists := e.keepaliveChs[key]; exists {
+		delete(e.keepaliveChs, key)
+		delete(e.keepaliveStop, key)
+	}
+	delete(e.leases, key)
+	delete(e.register, key)
+}
+
 // stopKeepAlive stops the keepalive goroutine for the given key
 func (e *etcdRegistry) stopKeepAlive(key string) {
 	e.Lock()

registry/etcd/etcd_test.go

@@ -269,3 +269,54 @@ func TestKeepAliveChannelReconnection(t *testing.T) {
 		t.Error("Stop channel should have been cleaned up after closure")
 	}
 }
+
+// TestHandleKeepAliveClosedClearsLease is a regression test for the bug
+// where a closed KeepAlive channel (e.g. because the lease expired on
+// the etcd server during a network partition) left the `leases` and
+// `register` caches populated, causing the next registerNode() to
+// short-circuit on the "unchanged" check and never re-register the
+// service. handleKeepAliveClosed must clear all four maps so that the
+// next heartbeat performs a full Grant+Put re-registration.
+func TestHandleKeepAliveClosedClearsLease(t *testing.T) {
+	reg := &etcdRegistry{
+		options:       registry.Options{Logger: logger.DefaultLogger},
+		register:      map[string]uint64{"svckey": 0xdeadbeef},
+		leases:        map[string]clientv3.LeaseID{"svckey": 42},
+		keepaliveChs:  map[string]<-chan *clientv3.LeaseKeepAliveResponse{"svckey": make(chan *clientv3.LeaseKeepAliveResponse)},
+		keepaliveStop: map[string]chan bool{"svckey": make(chan bool, 1)},
+	}
+
+	reg.handleKeepAliveClosed("svckey")
+
+	reg.RLock()
+	defer reg.RUnlock()
+
+	if _, ok := reg.keepaliveChs["svckey"]; ok {
+		t.Error("keepaliveChs not cleared")
+	}
+	if _, ok := reg.keepaliveStop["svckey"]; ok {
+		t.Error("keepaliveStop not cleared")
+	}
+	if _, ok := reg.leases["svckey"]; ok {
+		t.Error("leases not cleared — next heartbeat will skip re-registration")
+	}
+	if _, ok := reg.register["svckey"]; ok {
+		t.Error("register hash not cleared — next heartbeat will skip re-registration")
+	}
+}
+
+// TestHandleKeepAliveClosedIdempotent verifies that handleKeepAliveClosed
+// does not panic when called with a key that was already removed (which
+// can happen if stopKeepAlive/Deregister ran concurrently).
+func TestHandleKeepAliveClosedIdempotent(t *testing.T) {
+	reg := &etcdRegistry{
+		options:       registry.Options{Logger: logger.DefaultLogger},
+		register:      map[string]uint64{},
+		leases:        map[string]clientv3.LeaseID{},
+		keepaliveChs:  map[string]<-chan *clientv3.LeaseKeepAliveResponse{},
+		keepaliveStop: map[string]chan bool{},
+	}
+
+	// Should be a no-op; no panic.
+	reg.handleKeepAliveClosed("missing")
+}