diff --git a/pkg/manager/backendcluster/cluster.go b/pkg/manager/backendcluster/cluster.go index 748fd7484..1d4757a1e 100644 --- a/pkg/manager/backendcluster/cluster.go +++ b/pkg/manager/backendcluster/cluster.go @@ -127,13 +127,18 @@ func NewCluster( for key, query := range metricsQuerier.snapshot() { cluster.metrics.AddQueryExpr(key, query.expr, query.rule) } - if err := cluster.metrics.Start(ctx); err != nil { - _ = infoSyncer.Close() - if closeErr := etcdCli.Close(); closeErr != nil { - logger.Warn("close cluster etcd client failed after metrics init error", - zap.String("cluster", clusterCfg.Name), zap.Error(closeErr)) + // The Rust dataplane owns backend metric collection and its election endpoint. + // Keep the Go reader allocated so the manager/API surfaces remain safe, but do + // not let it campaign with api.addr, which is served by Rust in this mode. + if !cfg.RustDataplane.Enabled { + if err := cluster.metrics.Start(ctx); err != nil { + _ = infoSyncer.Close() + if closeErr := etcdCli.Close(); closeErr != nil { + logger.Warn("close cluster etcd client failed after metrics init error", + zap.String("cluster", clusterCfg.Name), zap.Error(closeErr)) + } + return nil, err } - return nil, err } return cluster, nil diff --git a/pkg/manager/backendcluster/manager_test.go b/pkg/manager/backendcluster/manager_test.go index 212843672..a3b6d9363 100644 --- a/pkg/manager/backendcluster/manager_test.go +++ b/pkg/manager/backendcluster/manager_test.go @@ -15,6 +15,7 @@ import ( "github.com/pingcap/tiproxy/lib/config" "github.com/pingcap/tiproxy/lib/util/logger" + "github.com/pingcap/tiproxy/pkg/balance/metricsreader" "github.com/pingcap/tiproxy/pkg/manager/infosync" "github.com/pingcap/tiproxy/pkg/testkit" "github.com/pingcap/tiproxy/pkg/util/etcd" @@ -68,6 +69,63 @@ func TestManagerFetchesAllClusters(t *testing.T) { }, 5*time.Second, 100*time.Millisecond) } +func TestManagerCampaignsForMetricsInGoDataplaneMode(t *testing.T) { + cluster := newManagerTestEtcdCluster(t) + t.Cleanup(func() { cluster.close(t) }) + + cfg := newManagerTestConfig() + cfg.Proxy.BackendClusters = []config.BackendCluster{ + {Name: "cluster-a", PDAddrs: cluster.addr}, + } + cfgGetter := newManagerTestConfigGetter(cfg) + cfgCh := make(chan *config.Config, 1) + + mgr := NewManager(zapLoggerForTest(t), nilClusterTLS) + require.NoError(t, mgr.Start(context.Background(), cfgGetter, cfgCh)) + t.Cleanup(func() { + close(cfgCh) + require.NoError(t, mgr.Close()) + }) + + const ownerPrefix = "/tiproxy/metric_reader/cluster-a/owner" + require.Eventually(t, func() bool { + response, err := cluster.kv.Get(context.Background(), ownerPrefix, clientv3.WithPrefix()) + return err == nil && response.Count > 0 + }, 5*time.Second, 25*time.Millisecond) +} + +func TestManagerDoesNotCampaignForMetricsInRustDataplaneMode(t *testing.T) { + cluster := newManagerTestEtcdCluster(t) + t.Cleanup(func() { cluster.close(t) }) + + cfg := newManagerTestConfig() + cfg.RustDataplane.Enabled = true + cfg.Proxy.BackendClusters = []config.BackendCluster{ + {Name: "cluster-a", PDAddrs: cluster.addr}, + } + cfgGetter := newManagerTestConfigGetter(cfg) + cfgCh := make(chan *config.Config, 1) + + mgr := NewManager(zapLoggerForTest(t), nilClusterTLS) + require.NoError(t, mgr.Start(context.Background(), cfgGetter, cfgCh)) + t.Cleanup(func() { + close(cfgCh) + require.NoError(t, mgr.Close()) + }) + + // Query registration and reads stay safe even though Go owns no collection + // lifecycle while the Rust dataplane is selected. + mgr.MetricsQuerier().AddQueryExpr("up", metricsreader.QueryExpr{PromQL: "up"}, metricsreader.QueryRule{}) + require.True(t, clusterHasBackendQueryRule(mgr.Snapshot()["cluster-a"], "up")) + require.Empty(t, mgr.MetricsQuerier().GetBackendMetricsByCluster("cluster-a")) + + const ownerPrefix = "/tiproxy/metric_reader/cluster-a/owner" + require.Never(t, func() bool { + response, err := cluster.kv.Get(context.Background(), ownerPrefix, clientv3.WithPrefix()) + return err != nil || response.Count != 0 + }, 500*time.Millisecond, 25*time.Millisecond) +} + func TestManagerDynamicClusterUpdate(t *testing.T) { clusterA := newManagerTestEtcdCluster(t) clusterB := newManagerTestEtcdCluster(t)