Implement KeyViz adapter labels

adapter/admin_grpc.go

@@ -699,6 +699,7 @@ func newKeyVizRowFrom(mr keyviz.MatrixRow, numCols int) *pb.KeyVizRow {
 	}
 	row := &pb.KeyVizRow{
 		BucketId:          bucketIDFor(mr),
+		Label:             string(mr.Label),
 		Start:             append([]byte(nil), mr.Start...),
 		End:               append([]byte(nil), mr.End...),
 		Aggregate:         mr.Aggregate,
@@ -722,6 +723,9 @@ func bucketIDFor(mr keyviz.MatrixRow) string {
 		return "virtual:" + strconv.FormatUint(mr.RouteID, 10)
 	}
 	id := "route:" + strconv.FormatUint(mr.RouteID, 10)
+	if mr.Label != keyviz.LabelLegacy {
+		id += ":" + string(mr.Label)
+	}
 	// Sub-bucket suffix only for genuinely sub-divided routes, so K=1 /
 	// aggregate / degenerate slots keep the exact legacy id. See §5.1.
 	if mr.SubBucketCount > 1 {

adapter/admin_grpc_subrange_test.go

@@ -37,6 +37,33 @@ func TestMatrixToProtoSubRowsDoNotCollide(t *testing.T) {
 	require.Equal(t, []uint64{9}, byID["route:1#1"])
 }
 
+func TestMatrixToProtoLabelsDoNotCollide(t *testing.T) {
+	t.Parallel()
+	pick := func(r keyviz.MatrixRow) uint64 { return r.Writes }
+	cols := []keyviz.MatrixColumn{
+		{
+			At: time.Unix(1_700_000_000, 0),
+			Rows: []keyviz.MatrixRow{
+				{RouteID: 1, Label: keyviz.LabelDynamo, Start: []byte("a"), End: []byte("z"), SubBucketCount: 1, Writes: 5},
+				{RouteID: 1, Label: keyviz.LabelRedis, Start: []byte("a"), End: []byte("z"), SubBucketCount: 1, Writes: 9},
+			},
+		},
+	}
+
+	resp := matrixToProto(cols, pick, 0)
+	require.Len(t, resp.Rows, 2)
+	byID := map[string]uint64{}
+	labels := map[string]string{}
+	for _, row := range resp.Rows {
+		byID[row.BucketId] = row.Values[0]
+		labels[row.BucketId] = row.Label
+	}
+	require.Equal(t, uint64(5), byID["route:1:dynamo"])
+	require.Equal(t, "dynamo", labels["route:1:dynamo"])
+	require.Equal(t, uint64(9), byID["route:1:redis"])
+	require.Equal(t, "redis", labels["route:1:redis"])
+}
+
 // TestNewKeyVizRowFromAggregateZeroTotalFallback pins the adapter/handler
 // harmonization (Claude round-2 follow-up): an aggregate row whose
 // MemberRoutesTotal is still 0 (a just-coalesced bucket serialized before

adapter/dynamodb.go

@@ -12,6 +12,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/bootjp/elastickv/keyviz"
 	"github.com/bootjp/elastickv/kv"
 	"github.com/bootjp/elastickv/monitoring"
 	"github.com/bootjp/elastickv/store"
@@ -259,7 +260,7 @@ func NewDynamoDBServer(listen net.Listener, st store.MVCCStore, coordinate kv.Co
 	d := &DynamoDBServer{
 		listen:           listen,
 		store:            st,
-		coordinator:      coordinate,
+		coordinator:      kv.WithKeyVizLabel(coordinate, keyviz.LabelDynamo),
 		onePhaseTxnDedup: os.Getenv("ELASTICKV_DYNAMODB_ONEPHASE_DEDUP") != "0",
 	}
 	d.targetHandlers = map[string]func(http.ResponseWriter, *http.Request){

adapter/grpc.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 
 	"github.com/bootjp/elastickv/internal"
+	"github.com/bootjp/elastickv/keyviz"
 	"github.com/bootjp/elastickv/kv"
 	pb "github.com/bootjp/elastickv/proto"
 	"github.com/bootjp/elastickv/store"
@@ -47,7 +48,7 @@ func NewGRPCServer(store store.MVCCStore, coordinate kv.Coordinator, opts ...GRP
 			Level: slog.LevelWarn,
 		})),
 		grpcTranscoder: newGrpcGrpcTranscoder(),
-		coordinator:    coordinate,
+		coordinator:    kv.WithKeyVizLabel(coordinate, keyviz.LabelRawKV),
 		store:          store,
 	}
 	for _, opt := range opts {

adapter/redis.go

@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"github.com/bootjp/elastickv/internal/raftengine"
+	"github.com/bootjp/elastickv/keyviz"
 	"github.com/bootjp/elastickv/kv"
 	"github.com/bootjp/elastickv/monitoring"
 	pb "github.com/bootjp/elastickv/proto"
@@ -476,7 +477,7 @@ func NewRedisServer(listen net.Listener, redisAddr string, store store.MVCCStore
 	r := &RedisServer{
 		listen:          listen,
 		store:           store,
-		coordinator:     coordinate,
+		coordinator:     kv.WithKeyVizLabel(coordinate, keyviz.LabelRedis),
 		redisTranscoder: newRedisTranscoder(),
 		redisAddr:       redisAddr,
 		relay:           relay,

adapter/redis_delta_compactor.go

@@ -7,6 +7,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/bootjp/elastickv/keyviz"
 	"github.com/bootjp/elastickv/kv"
 	"github.com/bootjp/elastickv/store"
 	"github.com/cockroachdb/errors"
@@ -113,7 +114,7 @@ func WithDeltaCompactorLogger(l *slog.Logger) DeltaCompactorOption {
 func NewDeltaCompactor(st store.MVCCStore, coord kv.Coordinator, opts ...DeltaCompactorOption) *DeltaCompactor {
 	c := &DeltaCompactor{
 		st:       st,
-		coord:    coord,
+		coord:    kv.WithKeyVizLabel(coord, keyviz.LabelRedis),
 		logger:   slog.Default(),
 		maxCount: defaultDeltaCompactorMaxDeltaCount,
 		interval: defaultDeltaCompactorScanInterval,

adapter/redis_delta_compactor_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/bootjp/elastickv/keyviz"
 	"github.com/bootjp/elastickv/kv"
 	"github.com/bootjp/elastickv/store"
 	"github.com/stretchr/testify/require"
@@ -20,6 +21,27 @@ func newDeltaCompactorTestFixture(t *testing.T) (store.MVCCStore, *DeltaCompacto
 	return st, c
 }
 
+type recordingCompactorCoordinator struct {
+	stubAdapterCoordinator
+	labels []keyviz.Label
+}
+
+func (c *recordingCompactorCoordinator) Dispatch(_ context.Context, reqs *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) {
+	if reqs != nil {
+		c.labels = append(c.labels, reqs.KeyVizLabel)
+	}
+	return &kv.CoordinateResponse{}, nil
+}
+
+func TestDeltaCompactorStampsRedisKeyVizLabel(t *testing.T) {
+	t.Parallel()
+	rec := &recordingCompactorCoordinator{}
+	c := NewDeltaCompactor(store.NewMVCCStore(), rec)
+	_, err := c.coord.Dispatch(context.Background(), &kv.OperationGroup[kv.OP]{})
+	require.NoError(t, err)
+	require.Equal(t, []keyviz.Label{keyviz.LabelRedis}, rec.labels)
+}
+
 func TestDeltaCompactor_ListDeltaFoldedIntoBaseMeta(t *testing.T) {
 	t.Parallel()
 

adapter/s3.go

@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/bootjp/elastickv/internal/s3keys"
+	"github.com/bootjp/elastickv/keyviz"
 	"github.com/bootjp/elastickv/kv"
 	"github.com/bootjp/elastickv/store"
 	"github.com/cockroachdb/errors"
@@ -318,7 +319,7 @@ func NewS3Server(listen net.Listener, s3Addr string, st store.MVCCStore, coordin
 		s3Addr:      s3Addr,
 		region:      s3DefaultRegion,
 		store:       st,
-		coordinator: coordinate,
+		coordinator: kv.WithKeyVizLabel(coordinate, keyviz.LabelS3),
 		leaderS3:    cloneLeaderAddrMap(leaderS3),
 		cleanupSem:  make(chan struct{}, s3ManifestCleanupWorkers),
 	}

adapter/sqs.go

@@ -10,6 +10,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/bootjp/elastickv/keyviz"
 	"github.com/bootjp/elastickv/kv"
 	"github.com/bootjp/elastickv/store"
 	"github.com/cockroachdb/errors"
@@ -285,7 +286,7 @@ func NewSQSServer(listen net.Listener, st store.MVCCStore, coordinate kv.Coordin
 	s := &SQSServer{
 		listen:       listen,
 		store:        st,
-		coordinator:  coordinate,
+		coordinator:  kv.WithKeyVizLabel(coordinate, keyviz.LabelSQS),
 		reaperCtx:    reaperCtx,
 		reaperCancel: reaperCancel,
 		throttle:     newBucketStoreDefault(),

docs/admin_ui_key_visualizer_design.md

@@ -135,7 +135,7 @@ Response matrix format: `matrix[i][j]` is the value for bucket `i` at time colum
 A single call site is added at the dispatch entry of `kv.ShardedCoordinator` (see `kv/sharded_coordinator.go`), immediately after the request is resolved to a `RouteID`:
 
 ```go
-sampler.Observe(routeID, op, keyLen, valueLen)
+sampler.Observe(routeID, key, op, valueLen, label)
 ```
 
 `sampler` is an interface; the default implementation is nil-safe (a nil sampler compiles to one branch and no allocation). The hook runs *before* Raft proposal so it measures offered load, not applied load.

docs/design/2026_05_25_proposed_keyviz_subrange_sampling.md

@@ -17,7 +17,7 @@ sequential prefix, change a schema). The current sampler cannot do
 this at the granularity the name implies.
 
 Today the sampler keys every counter on **RouteID**
-(`keyviz.MemSampler`, `Observe(routeID, op, keyLen, valueLen)` —
+(`keyviz.MemSampler`, `Observe(routeID, key, op, valueLen, label)` —
 `keyviz/sampler.go`). A *route* is the key range a Raft group owns in
 the distribution catalog. The heatmap's Y axis is therefore
 **route-granular**:
@@ -209,11 +209,12 @@ non-`MaxUint64` start does sub-divide.
 // before
 Observe(routeID uint64, op Op, keyLen, valueLen int)
 // after
-Observe(routeID uint64, key []byte, op Op, valueLen int)
+Observe(routeID uint64, key []byte, op Op, valueLen int, label Label)
 ```
 
 `keyLen` is dropped in favour of the key itself (`len(key)` recovers
-it for the `*Bytes` counters). This is a **semantic-touching signature
+it for the `*Bytes` counters), while adapter labels are forwarded
+unchanged by the sub-range bucket selection. This is a **semantic-touching signature
 change**: per CLAUDE.md, every caller of `Sampler.Observe` /
 `MemSampler.Observe` will be grep-audited so each passes the real key
 (not, say, a nil placeholder that would silently bucket all traffic

internal/admin/keyviz_fanout.go

@@ -551,6 +551,7 @@ type rowMergeAcc struct {
 	routeIDs          []uint64
 	routeIDsTruncated bool
 	routeCount        uint64
+	label             string
 	cells             []cellMergeAcc
 }
 
@@ -628,6 +629,7 @@ func mergeRowInto(
 			routeIDs:          append([]uint64(nil), row.RouteIDs...),
 			routeIDsTruncated: row.RouteIDsTruncated,
 			routeCount:        row.RouteCount,
+			label:             row.Label,
 			cells:             make([]cellMergeAcc, mergedWidth),
 		}
 		accByBucket[row.BucketID] = acc
@@ -745,6 +747,7 @@ func resolveRowMergeAcc(acc *rowMergeAcc, useGroupTermDedupe bool) KeyVizRow {
 	width := len(acc.cells)
 	row := KeyVizRow{
 		BucketID:          acc.bucketID,
+		Label:             acc.label,
 		Start:             acc.start,
 		End:               acc.end,
 		Aggregate:         acc.aggregate,

internal/admin/keyviz_fanout_test.go

@@ -43,6 +43,34 @@ func TestMergeKeyVizMatricesReadsSum(t *testing.T) {
 	require.False(t, merged.Rows[0].Conflict, "reads must never raise conflict")
 }
 
+func TestMergeKeyVizMatricesPreservesLabels(t *testing.T) {
+	t.Parallel()
+	col := []int64{1_700_000_000_000}
+	a := KeyVizMatrix{
+		ColumnUnixMs: col,
+		Series:       keyVizSeriesWrites,
+		Rows: []KeyVizRow{
+			{BucketID: "route:1:dynamo", Label: "dynamo", Values: []uint64{5}, RaftGroupIDs: []uint64{1}, LeaderTerms: []uint64{3}},
+		},
+	}
+	b := KeyVizMatrix{
+		ColumnUnixMs: col,
+		Series:       keyVizSeriesWrites,
+		Rows: []KeyVizRow{
+			{BucketID: "route:1:redis", Label: "redis", Values: []uint64{9}, RaftGroupIDs: []uint64{1}, LeaderTerms: []uint64{3}},
+		},
+	}
+
+	merged := mergeKeyVizMatrices([]KeyVizMatrix{a, b}, keyVizSeriesWrites)
+	require.Len(t, merged.Rows, 2)
+	byID := map[string]KeyVizRow{}
+	for _, row := range merged.Rows {
+		byID[row.BucketID] = row
+	}
+	require.Equal(t, "dynamo", byID["route:1:dynamo"].Label)
+	require.Equal(t, "redis", byID["route:1:redis"].Label)
+}
+
 // TestMergeKeyVizMatricesWritesMaxStableLeader pins the §4.2 happy
 // path: under stable leadership exactly one node reports non-zero
 // writes. The merge picks the non-zero value without raising

internal/admin/keyviz_handler.go

@@ -85,6 +85,7 @@ type KeyVizMatrix struct {
 // surfaces that per-cell conflict bit on the wire via Conflicts[].
 type KeyVizRow struct {
 	BucketID          string   `json:"bucket_id"`
+	Label             string   `json:"label,omitempty"`
 	Start             []byte   `json:"start"`
 	End               []byte   `json:"end"`
 	Aggregate         bool     `json:"aggregate"`
@@ -408,6 +409,7 @@ func newKeyVizRowFrom(mr keyviz.MatrixRow, numCols int) *KeyVizRow {
 	}
 	row := &KeyVizRow{
 		BucketID:          bucketIDFor(mr),
+		Label:             string(mr.Label),
 		Start:             append([]byte(nil), mr.Start...),
 		End:               append([]byte(nil), mr.End...),
 		Aggregate:         mr.Aggregate,
@@ -431,6 +433,9 @@ func bucketIDFor(mr keyviz.MatrixRow) string {
 		return "virtual:" + strconv.FormatUint(mr.RouteID, 10)
 	}
 	id := "route:" + strconv.FormatUint(mr.RouteID, 10)
+	if mr.Label != keyviz.LabelLegacy {
+		id += ":" + string(mr.Label)
+	}
 	// Append the sub-bucket index only for genuinely sub-divided routes
 	// (SubBucketCount > 1), so a K=1 / aggregate / degenerate slot keeps
 	// the exact legacy "route:<id>" id and the change stays inert at the

internal/admin/keyviz_hotkeys_fanout.go

@@ -258,6 +258,9 @@ func buildKeyVizHotKeysPeerURL(peer string, params hotKeysParams) (string, error
 	base.Path = pathKeyVizHotKeys
 	q := base.Query()
 	q.Set("route_id", strconv.FormatUint(params.routeID, 10))
+	if params.label != keyviz.LabelLegacy {
+		q.Set("label", string(params.label))
+	}
 	if params.subBucketSet {
 		q.Set("sub_bucket", strconv.Itoa(params.subBucket))
 	}
@@ -322,6 +325,7 @@ func mergeHotKeyResponses(responses []hotKeyResponse, params hotKeysParams) hotK
 		// cannot panic the merge step.
 		return hotKeyResponse{
 			RouteID:     params.routeID,
+			Label:       string(params.label),
 			Series:      params.series,
 			Approximate: true,
 			Keys:        []hotKeyResponseEntry{},
@@ -336,6 +340,7 @@ func mergeHotKeyResponses(responses []hotKeyResponse, params hotKeysParams) hotK
 	// the SUM identity and the loop folds responses[0] cleanly.
 	out := hotKeyResponse{
 		RouteID:     base.RouteID,
+		Label:       base.Label,
 		SubBucket:   base.SubBucket,
 		Series:      base.Series,
 		Approximate: true, // SS sketches are always approximate.

internal/admin/keyviz_hotkeys_fanout_test.go

@@ -259,6 +259,12 @@ func TestBuildKeyVizHotKeysPeerURL(t *testing.T) {
 			params: hotKeysParams{routeID: 7, series: "writes", top: 20},
 			want:   "https://node-b.internal:9443/admin/api/v1/keyviz/hotkeys?route_id=7&series=writes&top=256",
 		},
+		{
+			name:   "label forwarded",
+			peer:   "10.0.0.2:8080",
+			params: hotKeysParams{routeID: 7, label: keyviz.LabelRedis, series: "writes", top: 20},
+			want:   "http://10.0.0.2:8080/admin/api/v1/keyviz/hotkeys?label=redis&route_id=7&series=writes&top=256",
+		},
 		{
 			name: "sub_bucket + from/to forwarded",
 			peer: "10.0.0.2:8080",