keyspace: add optional keyspace info metric #11176
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -76,6 +76,7 @@ const ( | |
| // Config is the interface for keyspace config. | ||
| type Config interface { | ||
| GetPreAlloc() []string | ||
| IsKeyspaceLevelMetricsEnabled() bool | ||
| ToWaitRegionSplit() bool | ||
| GetWaitRegionSplitTimeout() time.Duration | ||
| GetCheckRegionSplitInterval() time.Duration | ||
|
|
@@ -242,7 +243,13 @@ func (manager *Manager) initReserveKeyspace(id uint32, name string) error { | |
|
|
||
| // UpdateConfig update keyspace manager's config. | ||
| func (manager *Manager) UpdateConfig(cfg Config) { | ||
| keyspaceInfoMetricsMu.Lock() | ||
| wasEnabled := manager.config.IsKeyspaceLevelMetricsEnabled() | ||
| manager.config = cfg | ||
| if wasEnabled && !cfg.IsKeyspaceLevelMetricsEnabled() { | ||
| resetKeyspaceInfoMetricsLocked() | ||
| } | ||
| keyspaceInfoMetricsMu.Unlock() | ||
| if manager.mgm != nil { | ||
| manager.mgm.updateGroups(cfg.GetMetaServiceGroups()) | ||
| } | ||
|
|
@@ -564,6 +571,7 @@ func (manager *Manager) saveNewKeyspace(keyspace *keyspacepb.KeyspaceMeta) error | |
| if err == nil { | ||
| // Update the keyspace name cache only after the transaction commits. | ||
| manager.keyspaceNameLookup.Store(keyspace.GetId(), keyspace.Name) | ||
| manager.UpdateKeyspaceInfoMetrics(keyspace) | ||
| } | ||
| return err | ||
| } | ||
|
|
@@ -770,6 +778,9 @@ func (manager *Manager) LoadKeyspace(name string) (*keyspacepb.KeyspaceMeta, err | |
| if manager.mgm != nil && meta != nil { | ||
| manager.mgm.AttachEndpoints(meta.GetConfig()) | ||
| } | ||
| if err == nil { | ||
| manager.UpdateKeyspaceInfoMetrics(meta) | ||
| } | ||
| return meta, err | ||
| } | ||
|
|
||
|
|
@@ -793,6 +804,9 @@ func (manager *Manager) LoadKeyspaceByID(spaceID uint32) (*keyspacepb.KeyspaceMe | |
| if manager.mgm != nil && meta != nil { | ||
| manager.mgm.AttachEndpoints(meta.GetConfig()) | ||
| } | ||
| if err == nil { | ||
| manager.UpdateKeyspaceInfoMetrics(meta) | ||
|
nolouch marked this conversation as resolved.
Outdated
Member
There was a problem hiding this comment. Blocking: a load that read before deletion can permanently recreate the deleted series. The read transaction finishes before this metric update. A concurrent The metric publication needs ordering or version validation against removal. A second unsynchronized existence check would leave the same check-then-write race. Unit-test reproducerAdd type blockingKeyspaceStorage struct {
endpoint.KeyspaceStorage
blockNext atomic.Bool
reached chan struct{}
release chan struct{}
}
func (s *blockingKeyspaceStorage) RunInTxn(ctx context.Context, f func(kv.Txn) error) error {
err := s.KeyspaceStorage.RunInTxn(ctx, f)
if s.blockNext.CompareAndSwap(true, false) {
close(s.reached)
<-s.release
}
return err
}
func (suite *keyspaceTestSuite) TestStaleLoadCannotRestoreRemovedInfoMetric() {
re := suite.Require()
suite.manager.UpdateConfig(&mockConfig{EnableKeyspaceLevelMetrics: true})
resetKeyspaceInfoMetrics()
created, err := suite.manager.CreateKeyspace(&CreateKeyspaceRequest{
Name: "review_stale_load", CreateTime: time.Now().Unix(),
})
re.NoError(err)
for _, state := range []keyspacepb.KeyspaceState{
keyspacepb.KeyspaceState_DISABLED,
keyspacepb.KeyspaceState_ARCHIVED,
keyspacepb.KeyspaceState_TOMBSTONE,
} {
_, err = suite.manager.UpdateKeyspaceState(created.GetName(), state, time.Now().Unix())
re.NoError(err)
}
store := &blockingKeyspaceStorage{
KeyspaceStorage: suite.manager.store,
reached: make(chan struct{}),
release: make(chan struct{}),
}
suite.manager.store = store
store.blockNext.Store(true)
loadDone := make(chan error, 1)
go func() {
_, loadErr := suite.manager.LoadKeyspaceByID(created.GetId())
loadDone <- loadErr
}()
<-store.reached
re.NoError(suite.manager.RemoveKeyspace(created.GetId()))
re.Equal(0, promtestutil.CollectAndCount(keyspaceInfo))
close(store.release)
re.NoError(<-loadDone)
re.Equal(0, promtestutil.CollectAndCount(keyspaceInfo))
}Run: On
Contributor
Author
There was a problem hiding this comment. This metric is now defined as a client-observed mapping, not a strongly consistent metadata snapshot. A concurrent deletion may leave a previously observed mapping temporarily, but deleted keyspaces won’t have active storage-size series to join with. Avoiding this window would require extra lifecycle synchronization, which we want to avoid here.
Member
There was a problem hiding this comment. The current head makes this broader than the concurrent window described above: no successful removal path deletes an observed series at all. func (suite *keyspaceTestSuite) TestRemovalClearsObservedInfoMetric() {
re := suite.Require()
resetKeyspaceInfoMetrics()
suite.T().Cleanup(resetKeyspaceInfoMetrics)
created, err := suite.manager.CreateKeyspace(&CreateKeyspaceRequest{
Name: "review_rm_metric", CreateTime: time.Now().Unix(),
})
re.NoError(err)
SetKeyspaceInfoMetrics(created.GetId(), created.GetName())
re.Equal(1, testutil.CollectAndCount(keyspaceInfo))
for _, state := range []keyspacepb.KeyspaceState{
keyspacepb.KeyspaceState_DISABLED,
keyspacepb.KeyspaceState_ARCHIVED,
} {
_, err = suite.manager.UpdateKeyspaceStateByID(
created.GetId(), state, time.Now().Unix())
re.NoError(err)
}
re.NoError(suite.manager.store.RunInTxn(suite.ctx, func(txn kv.Txn) error {
return suite.manager.RemoveKeyspace(txn, created.GetId())
}))
re.Equal(0, testutil.CollectAndCount(keyspaceInfo))
}Run: On This is explicitly required by Issue #11175. The fact that another storage-size metric may currently have no sample does not make stale high-cardinality labels safe: ID reuse makes the stale series ambiguous, and repeated create/remove cycles retain series indefinitely. Please restore post-commit cleanup. |
||
| } | ||
| return meta, err | ||
| } | ||
|
|
||
|
|
@@ -1037,31 +1051,59 @@ func (manager *Manager) UpdateKeyspaceState(name string, newState keyspacepb.Key | |
|
|
||
| // RemoveKeyspace removes the keyspace specified by id if it's in proper state and not protected. | ||
| func (manager *Manager) RemoveKeyspace(txn kv.Txn, id uint32) error { | ||
| _, err := manager.stageRemoveKeyspace(txn, id) | ||
| return err | ||
| } | ||
|
|
||
| // stageRemoveKeyspace stages keyspace removal and returns its metadata for post-commit cleanup. | ||
| func (manager *Manager) stageRemoveKeyspace(txn kv.Txn, id uint32) (*keyspacepb.KeyspaceMeta, error) { | ||
| manager.metaLock.Lock(id) | ||
| defer manager.metaLock.Unlock(id) | ||
| if isProtectedKeyspaceID(id) { | ||
| return nil, newModifyProtectedKeyspaceError() | ||
| } | ||
| meta, err := manager.store.LoadKeyspaceMeta(txn, id) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| if meta == nil { | ||
| return nil, errs.ErrKeyspaceNotFound | ||
| } | ||
| if meta.GetState() == keyspacepb.KeyspaceState_ENABLED || meta.GetState() == keyspacepb.KeyspaceState_DISABLED { | ||
| return nil, errors.Errorf("cannot remove keyspace in state %s", meta.GetState().String()) | ||
| } | ||
| err = manager.store.RemoveKeyspace(txn, id, meta.GetName()) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| // Keep the meta-service group assignment accounting in sync within the same | ||
| // txn. Without this, removed keyspaces leak count and could permanently block | ||
| // deleting an otherwise-empty group. | ||
| if err := manager.unassignKeyspaceFromMetaServiceGroup(txn, meta); err != nil { | ||
| return nil, err | ||
| } | ||
| return meta, nil | ||
| } | ||
|
|
||
| // finishRemoveKeyspace clears the caches and metrics after a keyspace removal transaction commits. | ||
| func (manager *Manager) finishRemoveKeyspace(meta *keyspacepb.KeyspaceMeta) { | ||
| manager.keyspaceNameLookup.Delete(meta.GetId()) | ||
| manager.keyspaceStateLookup.Delete(meta.GetId()) | ||
| keyspaceInfoMetricsMu.Lock() | ||
| defer keyspaceInfoMetricsMu.Unlock() | ||
| if manager.config.IsKeyspaceLevelMetricsEnabled() { | ||
| deleteKeyspaceInfoMetricsLocked(meta.GetId(), meta.GetName()) | ||
| } | ||
| } | ||
|
|
||
| // UpdateKeyspaceInfoMetrics updates the keyspace ID-to-name mapping metric when keyspace-level metrics are enabled. | ||
| func (manager *Manager) UpdateKeyspaceInfoMetrics(meta *keyspacepb.KeyspaceMeta) { | ||
| keyspaceInfoMetricsMu.Lock() | ||
| defer keyspaceInfoMetricsMu.Unlock() | ||
| if meta == nil || !manager.config.IsKeyspaceLevelMetricsEnabled() { | ||
|
nolouch marked this conversation as resolved.
Outdated
|
||
| return | ||
| } | ||
| setKeyspaceInfoMetricsLocked(meta.GetId(), meta.GetName()) | ||
| } | ||
|
|
||
| // UpdateKeyspaceStateByID updates target keyspace to the given state if it's not already in that state. | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -122,6 +122,7 @@ func (s *KeyspaceServer) WatchKeyspaces(request *keyspacepb.WatchKeyspacesReques | |
| defer cancel() // cancel context to stop watcher | ||
| return err | ||
| } | ||
| s.GetKeyspaceManager().UpdateKeyspaceInfoMetrics(meta) | ||
|
Member
There was a problem hiding this comment. Blocking: DELETE events never clear the metric. The Unit-test reproducerfunc TestReviewWatchDeleteRemovesMetric(t *testing.T) {
re := require.New(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
cluster, err := NewTestCluster(ctx, 1, func(c *config.Config, _ string) {
c.Keyspace.EnableKeyspaceLevelMetrics = true
})
re.NoError(err)
defer cluster.Destroy()
re.NoError(cluster.RunInitialServers())
leader := cluster.GetServer(cluster.WaitLeader())
re.NoError(leader.BootstrapCluster())
_, conn := testutil.MustNewGrpcClient(re, leader.GetAddr())
defer conn.Close()
watch, err := keyspacepb.NewKeyspaceClient(conn).WatchKeyspaces(ctx,
&keyspacepb.WatchKeyspacesRequest{
Header: &pdpb.RequestHeader{ClusterId: leader.GetClusterID()},
})
re.NoError(err)
_, err = watch.Recv() // initial snapshot
re.NoError(err)
put := func(id uint32, name string) {
raw, marshalErr := proto.Marshal(&keyspacepb.KeyspaceMeta{
Keyspace: &keyspacepb.KeyspaceMeta_Id{Id: id},
Name: name,
State: keyspacepb.KeyspaceState_ENABLED,
})
re.NoError(marshalErr)
_, putErr := leader.GetEtcdClient().Put(ctx, keypath.KeyspaceMetaPath(id), string(raw))
re.NoError(putErr)
}
put(1000, "review_watch_deleted")
testutil.Eventually(re, func() bool {
return metricExists(1000, "review_watch_deleted")
})
_, err = leader.GetEtcdClient().Delete(ctx, keypath.KeyspaceMetaPath(1000))
re.NoError(err)
put(1001, "review_watch_sentinel")
testutil.Eventually(re, func() bool {
return metricExists(1001, "review_watch_sentinel")
})
deleted := metricExists(1000, "review_watch_deleted")
cancel()
_ = conn.Close()
re.False(deleted)
}Here Run: Observed output: Please implement the delete callback and cover both an etcd DELETE event and the resulting metric removal.
Contributor
Author
There was a problem hiding this comment. Fixed in c02f996. WatchKeyspaces now parses the deleted keyspace ID and removes the series through the mutex-protected metric cache, which retains the corresponding name label. The cached-ID deletion path is covered by the keyspace metrics test.
Member
There was a problem hiding this comment. This has regressed on the current head. Commit I also ran the narrower test below to separate watch population from every create/load path: func TestWatchPutPopulatesKeyspaceInfoMetric(t *testing.T) {
re := require.New(t)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
cluster, err := NewTestCluster(ctx, 1, func(cfg *config.Config, _ string) {
cfg.Keyspace.EnableKeyspaceLevelMetrics = true
})
re.NoError(err)
defer cluster.Destroy()
re.NoError(cluster.RunInitialServers())
leader := cluster.GetServer(cluster.WaitLeader())
re.NoError(leader.BootstrapCluster())
_, conn := testutil.MustNewGrpcClient(re, leader.GetAddr())
defer conn.Close()
watch, err := keyspacepb.NewKeyspaceClient(conn).WatchKeyspaces(ctx,
&keyspacepb.WatchKeyspacesRequest{
Header: testutil.NewRequestHeader(leader.GetClusterID()),
})
re.NoError(err)
_, err = watch.Recv()
re.NoError(err)
const id, name = uint32(1000002), "review_watch_metric"
raw, err := proto.Marshal(&keyspacepb.KeyspaceMeta{
Keyspace: &keyspacepb.KeyspaceMeta_Id{Id: id},
Name: name,
State: keyspacepb.KeyspaceState_ENABLED,
})
re.NoError(err)
_, err = leader.GetEtcdClient().Put(ctx, keypath.KeyspaceMetaPath(id), string(raw))
re.NoError(err)
_, err = watch.Recv()
re.NoError(err)
re.True(metricExists(id, name))
}Run: On |
||
| keyspaces = append(keyspaces, meta) | ||
| return nil | ||
| } | ||
|
|
||
Uh oh!
There was an error while loading. Please reload this page.