-
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathhypercache_dist.go
More file actions
145 lines (121 loc) · 4 KB
/
hypercache_dist.go
File metadata and controls
145 lines (121 loc) · 4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
package hypercache
import (
"context"
"github.com/hyp3rd/hypercache/internal/eventbus"
"github.com/hyp3rd/hypercache/pkg/backend"
)
// EventBus returns the in-process broadcaster the distributed
// backend uses to publish topology changes (`members`,
// `heartbeat`). Returns nil when the backend is not DistMemory —
// the management HTTP server's SSE handler treats nil as
// "streaming unsupported" and falls back to a 503.
//
// The bus is owned by DistMemory and lives until the backend's
// lifecycle context is cancelled. SSE handlers Subscribe via the
// request context so they're reaped when either the client
// disconnects or the cache shuts down.
func (hyperCache *HyperCache[T]) EventBus() *eventbus.Bus {
if dm, ok := any(hyperCache.backend).(*backend.DistMemory); ok {
return dm.EventBus()
}
return nil
}
// DistMetrics returns distributed backend metrics if the underlying backend is DistMemory.
// Returns nil if unsupported.
func (hyperCache *HyperCache[T]) DistMetrics() any { // generic any to avoid exporting type into core interface
if dm, ok := any(hyperCache.backend).(*backend.DistMemory); ok {
m := dm.Metrics()
return m
}
return nil
}
// ClusterOwners returns the owners for a key if the distributed backend supports it; otherwise empty slice.
func (hyperCache *HyperCache[T]) ClusterOwners(key string) []string {
if dm, ok := any(hyperCache.backend).(*backend.DistMemory); ok {
owners := dm.DebugOwners(key)
out := make([]string, 0, len(owners))
for _, o := range owners {
out = append(out, string(o))
}
return out
}
return nil
}
// DistMembershipSnapshot returns a snapshot of membership if distributed backend; otherwise nil slice.
//
//nolint:nonamedreturns
func (hyperCache *HyperCache[T]) DistMembershipSnapshot() (members []struct {
ID string
Address string
State string
Incarnation uint64
}, replication, vnodes int,
) {
if dm, ok := any(hyperCache.backend).(*backend.DistMemory); ok {
membership := dm.Membership()
ring := dm.Ring()
if membership == nil || ring == nil {
return nil, 0, 0
}
nodes := membership.List()
out := make([]struct {
ID string
Address string
State string
Incarnation uint64
}, 0, len(nodes))
for _, node := range nodes {
out = append(out, struct {
ID string
Address string
State string
Incarnation uint64
}{
ID: string(node.ID),
Address: node.Address,
State: node.State.String(),
Incarnation: node.Incarnation,
})
}
return out, ring.Replication(), ring.VirtualNodesPerNode()
}
return nil, 0, 0
}
// DistRingHashSpots returns vnode hashes as hex strings if available (debug).
func (hyperCache *HyperCache[T]) DistRingHashSpots() []string {
if dm, ok := any(hyperCache.backend).(*backend.DistMemory); ok {
if ring := dm.Ring(); ring != nil {
return ring.VNodeHashes()
}
}
return nil
}
// DistDrain marks the underlying distributed backend for graceful
// shutdown when one is configured: /health flips to 503 on the dist
// HTTP listener, Set/Remove return sentinel.ErrDraining, Get
// continues to serve. Returns nil when the backend is not a
// DistMemory (no-op for in-memory and Redis backends), so callers
// don't need to type-check before invoking it.
//
// One-way and idempotent. Operators clear it by restarting the
// process after Drain settles and the cache has been Stopped.
func (hyperCache *HyperCache[T]) DistDrain(ctx context.Context) error {
dm, ok := any(hyperCache.backend).(*backend.DistMemory)
if !ok {
return nil
}
return dm.Drain(ctx)
}
// DistHeartbeatMetrics returns distributed heartbeat metrics if supported.
func (hyperCache *HyperCache[T]) DistHeartbeatMetrics() any {
if dm, ok := any(hyperCache.backend).(*backend.DistMemory); ok {
m := dm.Metrics()
return map[string]any{
"heartbeatSuccess": m.HeartbeatSuccess,
"heartbeatFailure": m.HeartbeatFailure,
"nodesRemoved": m.NodesRemoved,
"readPrimaryPromote": m.ReadPrimaryPromote,
}
}
return nil
}