diff --git a/api/core/v1alpha1/semver.go b/api/core/v1alpha1/semver.go index 1503ce5b..43305f17 100644 --- a/api/core/v1alpha1/semver.go +++ b/api/core/v1alpha1/semver.go @@ -28,19 +28,41 @@ const ( ) var ( + // featureVersions lists the minimum version per major branch from which the feature is + // available. versionPrecedes() uses major-version isolation, so every new major branch that + // should retain a feature needs its own explicit entry. Do NOT bulk-add a new major version + // to every feature "for consistency" — each feature must be verified independently against + // the new major branch before its gate is extended, to avoid unintentionally flipping + // behavior (e.g. lock migration handshake, pipeline/sharding stats collection). featureVersions = map[MOFeature][]semver.Version{ MOFeaturePipelineInfo: {semver.MustParse("1.1.2"), semver.MustParse("1.2.0"), semver.MustParse("2.0.0")}, MOFeatureSessionSource: {semver.MustParse("1.1.2"), semver.MustParse("1.2.0"), semver.MustParse("2.0.0")}, MOFeatureLockMigration: {semver.MustParse("1.2.0"), semver.MustParse("2.0.0")}, MOFeatureShardingMigration: {semver.MustParse("2.0.0")}, - MOFeatureDiscoveryFixed: {semver.MustParse("2.0.0")}, + } + + // featureGlobalMinVersions lists features that are stable across all future major versions + // once introduced. Unlike featureVersions, these do NOT need a new entry per major version + // because the underlying mechanism is a stable config field / protocol that MO guarantees + // to keep indefinitely. Only add features here when you are confident they will never be + // removed or incompatibly changed in future major versions. + featureGlobalMinVersions = map[MOFeature]semver.Version{ + // discovery-address is a stable [hakeeper-client] toml field supported since MO 2.0. + // It relies on K8s Service routing (operator-side), not on any MO-internal protocol + // that could change between major versions, so no per-major re-verification is needed. + MOFeatureDiscoveryFixed: semver.MustParse("2.0.0"), } MinimalVersion = semver.Version{Major: 0, Minor: 0, Patch: 0} ) -// HasMOFeature returns whether a version contains certain MO feature +// HasMOFeature returns whether a version contains certain MO feature. +// It checks featureGlobalMinVersions first (cross-major stable features), then +// featureVersions (per-major-verified features). func HasMOFeature(v semver.Version, f MOFeature) bool { + if minVer, ok := featureGlobalMinVersions[f]; ok && versionPrecedesCrossMajor(minVer, v) { + return true + } for _, minVersion := range featureVersions[f] { if versionPrecedes(minVersion, v) { return true @@ -49,8 +71,10 @@ func HasMOFeature(v semver.Version, f MOFeature) bool { return false } -// versionPrecedes returns whether current version is a strict preceding version of base version. -// for example, 1.2.1 is a strict preceding version of 1.1.0, but not 1.1.1 +// versionPrecedes returns whether current version is a strict preceding version of base version +// within the same major. Different major versions have no preceding relationship here, so every +// new major branch requires its own explicit entry in featureVersions. +// Example: 1.2.1 precedes 1.1.0, but 2.1.0 does NOT precede 1.1.0. func versionPrecedes(baseVersion semver.Version, current semver.Version) bool { if baseVersion.Major != current.Major { // different major version has no preceding relationship @@ -62,3 +86,16 @@ func versionPrecedes(baseVersion semver.Version, current semver.Version) bool { } return baseVersion.Minor == current.Minor && current.Patch >= baseVersion.Patch } + +// versionPrecedesCrossMajor is like versionPrecedes but without major-version isolation. +// Use this only for stable config fields / protocols guaranteed never to be removed across +// future major versions (see featureGlobalMinVersions). +func versionPrecedesCrossMajor(baseVersion semver.Version, current semver.Version) bool { + if current.Major != baseVersion.Major { + return current.Major > baseVersion.Major + } + if baseVersion.Patch == 0 && current.Minor >= baseVersion.Minor { + return true + } + return baseVersion.Minor == current.Minor && current.Patch >= baseVersion.Patch +} diff --git a/api/core/v1alpha1/semver_test.go b/api/core/v1alpha1/semver_test.go index 001d31cf..854a1a4f 100644 --- a/api/core/v1alpha1/semver_test.go +++ b/api/core/v1alpha1/semver_test.go @@ -22,6 +22,19 @@ import ( . "github.com/onsi/gomega" ) +// TestFeatureMapsNoOverlap guards against accidentally placing the same feature in both +// featureGlobalMinVersions and featureVersions. If a feature appears in both maps, +// HasMOFeature() silently takes the global path and the per-major entries become dead code, +// making the behavior hard to reason about. +func TestFeatureMapsNoOverlap(t *testing.T) { + g := NewGomegaWithT(t) + for f := range featureGlobalMinVersions { + _, inPerMajor := featureVersions[f] + g.Expect(inPerMajor).To(BeFalse(), + "feature %q is defined in both featureGlobalMinVersions and featureVersions; pick one", f) + } +} + func TestHasMOFeature(t *testing.T) { g := NewGomegaWithT(t) g.Expect(HasMOFeature(mustParse("1.1.2"), MOFeaturePipelineInfo)).To(BeTrue()) @@ -37,10 +50,52 @@ func TestHasMOFeature(t *testing.T) { g.Expect(HasMOFeature(mustParse("v1.2.2-woraround-something-else"), MOFeatureLockMigration)).To(BeTrue()) g.Expect(HasMOFeature(mustParse("2.0.1"), MOFeatureLockMigration)).To(BeTrue()) featureVersions["dummy"] = []semver.Version{mustParse("1.2.3")} + t.Cleanup(func() { delete(featureVersions, "dummy") }) g.Expect(HasMOFeature(mustParse("v1.2.3"), "dummy")).To(BeTrue()) g.Expect(HasMOFeature(mustParse("v1.3.0"), "dummy")).To(BeFalse()) } +// TestHasMOFeature_DiscoveryFixed is a regression test for issue #597. +// MOFeatureDiscoveryFixed is now in featureGlobalMinVersions (cross-major), so it must return +// true for all MO versions >= 2.0.0 regardless of major — including future 4.x, 5.x, etc. — +// without needing a new entry per major version. +func TestHasMOFeature_DiscoveryFixed(t *testing.T) { + g := NewGomegaWithT(t) + // MO 2.x — original fix + g.Expect(HasMOFeature(mustParse("2.0.0"), MOFeatureDiscoveryFixed)).To(BeTrue()) + g.Expect(HasMOFeature(mustParse("2.1.0"), MOFeatureDiscoveryFixed)).To(BeTrue()) + // MO 3.x — regression from issue #597 + g.Expect(HasMOFeature(mustParse("3.0.0"), MOFeatureDiscoveryFixed)).To(BeTrue()) + g.Expect(HasMOFeature(mustParse("3.0.16"), MOFeatureDiscoveryFixed)).To(BeTrue()) + g.Expect(HasMOFeature(mustParse("v3.0.16-bda2d138a-2026-06-24"), MOFeatureDiscoveryFixed)).To(BeTrue()) + // MO 4.x — must work without any new entry in featureGlobalMinVersions + g.Expect(HasMOFeature(mustParse("4.0.0"), MOFeatureDiscoveryFixed)).To(BeTrue()) + g.Expect(HasMOFeature(mustParse("4.5.2"), MOFeatureDiscoveryFixed)).To(BeTrue()) + // MO 1.x — discovery-address not yet supported + g.Expect(HasMOFeature(mustParse("1.2.0"), MOFeatureDiscoveryFixed)).To(BeFalse()) + g.Expect(HasMOFeature(mustParse("1.9.9"), MOFeatureDiscoveryFixed)).To(BeFalse()) +} + +// TestHasMOFeature_OtherFeaturesNotExtendedTo3x guards against accidentally widening the +// version gate for features that have NOT been explicitly verified against MO 3.x. Extending +// featureVersions in bulk (i.e. blindly adding "3.0.0" to every feature) would silently flip +// unrelated behavior (lock migration handshake, pipeline/sharding stats collection, session +// source accounting) on MO 3.x without dedicated verification. Only MOFeatureDiscoveryFixed +// has been confirmed compatible with 3.x so far (see #597); this test should be updated +// deliberately, one feature at a time, as each is verified. +func TestHasMOFeature_OtherFeaturesNotExtendedTo3x(t *testing.T) { + g := NewGomegaWithT(t) + unverifiedOn3x := []MOFeature{ + MOFeaturePipelineInfo, + MOFeatureSessionSource, + MOFeatureLockMigration, + MOFeatureShardingMigration, + } + for _, f := range unverifiedOn3x { + g.Expect(HasMOFeature(mustParse("3.0.0"), f)).To(BeFalse(), "feature %s should not yet be enabled on MO 3.x", f) + } +} + func mustParse(s string) semver.Version { v, err := semver.ParseTolerant(s) if err != nil { diff --git a/pkg/controllers/cnset/controller.go b/pkg/controllers/cnset/controller.go index 6e25c2fc..dd92db3e 100644 --- a/pkg/controllers/cnset/controller.go +++ b/pkg/controllers/cnset/controller.go @@ -19,9 +19,11 @@ import ( "time" "github.com/matrixorigin/matrixone-operator/api/features" + "github.com/matrixorigin/matrixone-operator/pkg/controllers/logset" "github.com/matrixorigin/matrixone-operator/pkg/utils" "github.com/openkruise/kruise-api/apps/pub" kruisev1alpha1 "github.com/openkruise/kruise-api/apps/v1alpha1" + kruise "github.com/openkruise/kruise-api/apps/v1beta1" "k8s.io/utils/pointer" "github.com/go-errors/errors" @@ -351,7 +353,18 @@ func syncCloneSet(ctx *recon.Context[*v1alpha1.CNSet], cs *kruisev1alpha1.CloneS } } - cm, configSuffix, err := buildCNSetConfigMap(ctx.Obj, ctx.Dep.Deps.LogSet) + // reservedOrdinals is only used in the service-addresses branch of buildCNSetConfigMap. + // When MOFeatureDiscoveryFixed is enabled the branch is never reached, so skip the + // extra STS GET to avoid an unnecessary dependency and potential requeue on transient errors. + var reservedOrdinals []int + if sv, ok := cn.Spec.GetSemVer(); !ok || !v1alpha1.HasMOFeature(*sv, v1alpha1.MOFeatureDiscoveryFixed) { + var err error + if reservedOrdinals, err = fetchLogSetReservedOrdinals(ctx, ctx.Dep.Deps.LogSet); err != nil { + return errors.WrapPrefix(err, "fetch logset reserved ordinals", 0) + } + } + + cm, configSuffix, err := buildCNSetConfigMap(ctx.Obj, ctx.Dep.Deps.LogSet, reservedOrdinals) if err != nil { return err } @@ -361,6 +374,26 @@ func syncCloneSet(ctx *recon.Context[*v1alpha1.CNSet], cs *kruisev1alpha1.CloneS return common.SyncConfigMap(ctx, &cs.Spec.Template.Spec, cm, cn.Spec.GetOperatorVersion()) } +// fetchLogSetReservedOrdinals fetches the kruise StatefulSet that backs the given LogSet and +// returns its spec.reserveOrdinals list. This allows CN config builders to generate +// accurate service-addresses that skip ordinal holes created during LogService failover (#596). +// +// Any error (including "not found") is propagated to the caller instead of being swallowed: +// by the time CN builds its ConfigMap, ls.Status.Discovery is already required to be set, +// which implies the LogSet (and its StatefulSet) must exist. Silently falling back to "no +// holes" on a transient read error could regenerate a service-addresses list that still +// points at a dead ordinal, defeating the purpose of this fix. Reconcile should simply retry. +func fetchLogSetReservedOrdinals(ctx *recon.Context[*v1alpha1.CNSet], ls *v1alpha1.LogSet) ([]int, error) { + if ls == nil { + return nil, nil + } + sts := &kruise.StatefulSet{} + if err := ctx.Get(client.ObjectKey{Namespace: ls.Namespace, Name: logset.LogSetStsName(ls)}, sts); err != nil { + return nil, err + } + return sts.Spec.ReserveOrdinals, nil +} + func setReady(cn *v1alpha1.CNSet) { cn.Status.SetCondition(metav1.Condition{ Type: recon.ConditionTypeReady, diff --git a/pkg/controllers/cnset/controller_test.go b/pkg/controllers/cnset/controller_test.go index bb29dc33..279442ae 100644 --- a/pkg/controllers/cnset/controller_test.go +++ b/pkg/controllers/cnset/controller_test.go @@ -155,6 +155,14 @@ func TestCNSetActor_Observe(t *testing.T) { Type: corev1.ServiceTypeLoadBalancer, }, }, + // the LogSet's own StatefulSet must exist by the time CN builds its + // ConfigMap (fetchLogSetReservedOrdinals requires it, see #596). + &kruisev1.StatefulSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-log", + Namespace: "default", + }, + }, ).Build(), }, expect: func(g *WithT, action recon.Action[*v1alpha1.CNSet], cli client.Client, err error) { @@ -300,6 +308,74 @@ func TestCNSetVolumeMount(t *testing.T) { } } +// Test_fetchLogSetReservedOrdinals is a regression test for issue #596: previously any +// error reading the LogSet StatefulSet (including "not found") was swallowed and treated +// as "no ordinal holes", which could cause service-addresses to be regenerated with a dead +// ordinal on a transient read failure. Errors must now propagate so reconcile retries. +func Test_fetchLogSetReservedOrdinals(t *testing.T) { + s := newScheme() + cn := &v1alpha1.CNSet{ + ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "test"}, + } + ls := &v1alpha1.LogSet{ + ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "test"}, + } + + tests := []struct { + name string + client client.Client + ls *v1alpha1.LogSet + wantOrdinals []int + wantErr bool + }{ + { + name: "sts exists with reserved ordinals", + client: &fake.Client{ + Client: fake.KubeClientBuilder().WithScheme(s).WithObjects( + &kruisev1.StatefulSet{ + ObjectMeta: metav1.ObjectMeta{Name: "test-log", Namespace: "default"}, + Spec: kruisev1.StatefulSetSpec{ReserveOrdinals: []int{1}}, + }, + ).Build(), + }, + ls: ls, + wantOrdinals: []int{1}, + }, + { + name: "sts not found propagates error instead of falling back silently", + client: &fake.Client{ + Client: fake.KubeClientBuilder().WithScheme(s).Build(), + }, + ls: ls, + wantErr: true, + }, + { + name: "nil logset returns no ordinals and no error", + client: &fake.Client{ + Client: fake.KubeClientBuilder().WithScheme(s).Build(), + }, + ls: nil, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + g := NewGomegaWithT(t) + mockCtrl := gomock.NewController(t) + eventEmitter := fake.NewMockEventEmitter(mockCtrl) + ctx := fake.NewContext(cn, tt.client, eventEmitter) + + got, err := fetchLogSetReservedOrdinals(ctx, tt.ls) + if tt.wantErr { + g.Expect(err).NotTo(BeNil()) + } else { + g.Expect(err).To(BeNil()) + g.Expect(got).To(Equal(tt.wantOrdinals)) + } + }) + } +} + func newScheme() *runtime.Scheme { scheme := runtime.NewScheme() utilruntime.Must(clientgoscheme.AddToScheme(scheme)) diff --git a/pkg/controllers/cnset/resource.go b/pkg/controllers/cnset/resource.go index eb7e2ed2..dfa2d8b3 100644 --- a/pkg/controllers/cnset/resource.go +++ b/pkg/controllers/cnset/resource.go @@ -271,7 +271,10 @@ func syncPodSpec(cn *v1alpha1.CNSet, cs *kruisev1alpha1.CloneSet, sp v1alpha1.Sh } } -func buildCNSetConfigMap(cn *v1alpha1.CNSet, ls *v1alpha1.LogSet) (*corev1.ConfigMap, string, error) { +// buildCNSetConfigMap builds the ConfigMap for a CNSet. +// reservedOrdinals should be set to the LogSet StatefulSet's spec.reserveOrdinals so that +// service-addresses correctly skips ordinal holes created during failover (issue #596). +func buildCNSetConfigMap(cn *v1alpha1.CNSet, ls *v1alpha1.LogSet, reservedOrdinals []int) (*corev1.ConfigMap, string, error) { if ls.Status.Discovery == nil { return nil, "", errors.New("logset had not yet exposed HAKeeper discovery address") } @@ -286,7 +289,7 @@ func buildCNSetConfigMap(cn *v1alpha1.CNSet, ls *v1alpha1.LogSet) (*corev1.Confi // via discovery-address, operator can take off unhealthy logstores without restart CN/TN cfg.Set([]string{"hakeeper-client", "discovery-address"}, ls.Status.Discovery.String()) } else { - cfg.Set([]string{"hakeeper-client", "service-addresses"}, logset.HaKeeperAdds(ls)) + cfg.Set([]string{"hakeeper-client", "service-addresses"}, logset.HaKeeperSvcAddrs(ls, reservedOrdinals)) } cfg.Set([]string{"cn", "role"}, cn.Spec.Role) cfg.Set([]string{"cn", "lockservice", "listen-address"}, fmt.Sprintf("0.0.0.0:%d", common.LockServicePort)) diff --git a/pkg/controllers/cnset/resource_test.go b/pkg/controllers/cnset/resource_test.go index 47a48e2d..4f2a909a 100644 --- a/pkg/controllers/cnset/resource_test.go +++ b/pkg/controllers/cnset/resource_test.go @@ -26,8 +26,9 @@ import ( func Test_buildCNSetConfigMap(t *testing.T) { type args struct { - cn *v1alpha1.CNSet - ls *v1alpha1.LogSet + cn *v1alpha1.CNSet + ls *v1alpha1.LogSet + reservedOrdinals []int } tests := []struct { name string @@ -221,15 +222,81 @@ memory-capacity = "1B" [hakeeper-client] service-addresses = [] +`, + }, + { + // regression test for #596: reservedOrdinals must be threaded through to + // HaKeeperSvcAddrs so that service-addresses skips the failed ordinal and + // includes the newly created replacement pod. + name: "1.x failover skips reserved ordinal", + args: args{ + cn: &v1alpha1.CNSet{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "test", + Name: "test", + }, + }, + ls: &v1alpha1.LogSet{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "test", + Name: "test", + }, + Spec: v1alpha1.LogSetSpec{ + PodSet: v1alpha1.PodSet{Replicas: 3}, + SharedStorage: v1alpha1.SharedStorageProvider{ + FileSystem: &v1alpha1.FileSystemProvider{ + Path: "/test", + }, + }, + }, + Status: v1alpha1.LogSetStatus{ + Discovery: &v1alpha1.LogSetDiscovery{ + Port: 6001, + Address: "test", + }, + }, + }, + reservedOrdinals: []int{1}, + }, + wantConfig: `data-dir = "/var/lib/matrixone/data" +service-type = "CN" + +[cn] +port-base = 6002 +role = "" + +[cn.lockservice] +listen-address = "0.0.0.0:6003" + +[[fileservice]] +backend = "DISK" +data-dir = "/var/lib/matrixone/data" +name = "LOCAL" + +[[fileservice]] +backend = "DISK" +data-dir = "/test" +name = "S3" + +[[fileservice]] +backend = "DISK-ETL" +data-dir = "/test" +name = "ETL" + +[fileservice.cache] +memory-capacity = "1B" + +[hakeeper-client] +service-addresses = ["test-log-0.test-log-headless.test.svc:32001", "test-log-2.test-log-headless.test.svc:32001", "test-log-3.test-log-headless.test.svc:32001"] `, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { g := NewGomegaWithT(t) - got, configSuffix, err := buildCNSetConfigMap(tt.args.cn, tt.args.ls) + got, configSuffix, err := buildCNSetConfigMap(tt.args.cn, tt.args.ls, tt.args.reservedOrdinals) if (err != nil) != tt.wantErr { - t.Errorf("buildDNSetConfigMap() error = %v, wantErr %v", err, tt.wantErr) + t.Errorf("buildCNSetConfigMap() error = %v, wantErr %v", err, tt.wantErr) return } configKey := "config.toml" diff --git a/pkg/controllers/dnset/controller.go b/pkg/controllers/dnset/controller.go index 0b1de64b..927ed308 100644 --- a/pkg/controllers/dnset/controller.go +++ b/pkg/controllers/dnset/controller.go @@ -19,6 +19,7 @@ import ( "time" "github.com/matrixorigin/matrixone-operator/api/features" + "github.com/matrixorigin/matrixone-operator/pkg/controllers/logset" "github.com/matrixorigin/matrixone-operator/pkg/utils" "github.com/go-errors/errors" @@ -120,8 +121,18 @@ func (d *Actor) Observe(ctx *recon.Context[*v1alpha1.DNSet]) (recon.Action[*v1al return d.with(sts, svc).Scale, nil } + // reservedOrdinals is only used in the service-addresses branch of buildDNSetConfigMap. + // When MOFeatureDiscoveryFixed is enabled the branch is never reached, so skip the + // extra STS GET to avoid an unnecessary dependency and potential requeue on transient errors. + var reservedOrdinals []int + if sv, ok := dn.Spec.GetSemVer(); !ok || !v1alpha1.HasMOFeature(*sv, v1alpha1.MOFeatureDiscoveryFixed) { + if reservedOrdinals, err = fetchLogSetReservedOrdinals(ctx, ctx.Dep.Deps.LogSet); err != nil { + return nil, errors.WrapPrefix(err, "fetch logset reserved ordinals", 0) + } + } + origin := sts.DeepCopy() - if err := syncPods(ctx, sts); err != nil { + if err := syncPods(ctx, sts, reservedOrdinals); err != nil { return nil, err } @@ -192,7 +203,15 @@ func (d *Actor) Create(ctx *recon.Context[*v1alpha1.DNSet]) error { syncPodSpec(dn, dnSet, ctx.Dep.Deps.LogSet.Spec.SharedStorage) syncPersistentVolumeClaim(dn, dnSet) - configMap, configSuffix, err := buildDNSetConfigMap(dn, ctx.Dep.Deps.LogSet) + var reservedOrdinals []int + if sv, ok := dn.Spec.GetSemVer(); !ok || !v1alpha1.HasMOFeature(*sv, v1alpha1.MOFeatureDiscoveryFixed) { + var err error + if reservedOrdinals, err = fetchLogSetReservedOrdinals(ctx, ctx.Dep.Deps.LogSet); err != nil { + return errors.WrapPrefix(err, "fetch logset reserved ordinals", 0) + } + } + + configMap, configSuffix, err := buildDNSetConfigMap(dn, ctx.Dep.Deps.LogSet, reservedOrdinals) if err != nil { return err } @@ -258,6 +277,26 @@ func (d *Actor) syncMetricService(ctx *recon.Context[*v1alpha1.DNSet]) error { }) } +// fetchLogSetReservedOrdinals fetches the kruise StatefulSet that backs the given LogSet and +// returns its spec.reserveOrdinals list. This allows DN/CN config builders to generate +// accurate service-addresses that skip ordinal holes created during LogService failover (#596). +// +// Any error (including "not found") is propagated to the caller instead of being swallowed: +// by the time DN/CN build their ConfigMap, ls.Status.Discovery is already required to be set, +// which implies the LogSet (and its StatefulSet) must exist. Silently falling back to "no +// holes" on a transient read error could regenerate a service-addresses list that still +// points at a dead ordinal, defeating the purpose of this fix. Reconcile should simply retry. +func fetchLogSetReservedOrdinals(ctx *recon.Context[*v1alpha1.DNSet], ls *v1alpha1.LogSet) ([]int, error) { + if ls == nil { + return nil, nil + } + sts := &kruise.StatefulSet{} + if err := ctx.Get(client.ObjectKey{Namespace: ls.Namespace, Name: logset.LogSetStsName(ls)}, sts); err != nil { + return nil, err + } + return sts.Spec.ReserveOrdinals, nil +} + func (d *Actor) Reconcile(mgr manager.Manager) error { err := recon.Setup[*v1alpha1.DNSet](&v1alpha1.DNSet{}, "dnset", mgr, d, recon.WithBuildFn(func(b *builder.Builder) { diff --git a/pkg/controllers/dnset/controller_test.go b/pkg/controllers/dnset/controller_test.go index fcca0a55..de49d56d 100644 --- a/pkg/controllers/dnset/controller_test.go +++ b/pkg/controllers/dnset/controller_test.go @@ -149,6 +149,14 @@ func TestDNSetActor_Observe(t *testing.T) { Namespace: "default", }, }, + // the LogSet's own StatefulSet must exist by the time DN builds its + // ConfigMap (fetchLogSetReservedOrdinals requires it, see #596). + &kruisev1.StatefulSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: "test-log", + Namespace: "default", + }, + }, ).Build(), }, expect: func(g *WithT, action recon.Action[*v1alpha1.DNSet], err error) { @@ -266,6 +274,74 @@ func TestDNSetVolumeMount(t *testing.T) { } } +// Test_fetchLogSetReservedOrdinals is a regression test for issue #596: previously any +// error reading the LogSet StatefulSet (including "not found") was swallowed and treated +// as "no ordinal holes", which could cause service-addresses to be regenerated with a dead +// ordinal on a transient read failure. Errors must now propagate so reconcile retries. +func Test_fetchLogSetReservedOrdinals(t *testing.T) { + s := newScheme() + dn := &v1alpha1.DNSet{ + ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "test"}, + } + ls := &v1alpha1.LogSet{ + ObjectMeta: metav1.ObjectMeta{Namespace: "default", Name: "test"}, + } + + tests := []struct { + name string + client client.Client + ls *v1alpha1.LogSet + wantOrdinals []int + wantErr bool + }{ + { + name: "sts exists with reserved ordinals", + client: &fake.Client{ + Client: fake.KubeClientBuilder().WithScheme(s).WithObjects( + &kruisev1.StatefulSet{ + ObjectMeta: metav1.ObjectMeta{Name: "test-log", Namespace: "default"}, + Spec: kruisev1.StatefulSetSpec{ReserveOrdinals: []int{1}}, + }, + ).Build(), + }, + ls: ls, + wantOrdinals: []int{1}, + }, + { + name: "sts not found propagates error instead of falling back silently", + client: &fake.Client{ + Client: fake.KubeClientBuilder().WithScheme(s).Build(), + }, + ls: ls, + wantErr: true, + }, + { + name: "nil logset returns no ordinals and no error", + client: &fake.Client{ + Client: fake.KubeClientBuilder().WithScheme(s).Build(), + }, + ls: nil, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + g := NewGomegaWithT(t) + mockCtrl := gomock.NewController(t) + eventEmitter := fake.NewMockEventEmitter(mockCtrl) + ctx := fake.NewContext(dn, tt.client, eventEmitter) + + got, err := fetchLogSetReservedOrdinals(ctx, tt.ls) + if tt.wantErr { + g.Expect(err).NotTo(BeNil()) + } else { + g.Expect(err).To(BeNil()) + g.Expect(got).To(Equal(tt.wantOrdinals)) + } + }) + } +} + func newScheme() *runtime.Scheme { scheme := runtime.NewScheme() utilruntime.Must(clientgoscheme.AddToScheme(scheme)) diff --git a/pkg/controllers/dnset/resource.go b/pkg/controllers/dnset/resource.go index 30e903c0..6322e9a8 100644 --- a/pkg/controllers/dnset/resource.go +++ b/pkg/controllers/dnset/resource.go @@ -198,8 +198,10 @@ func syncPodSpec(dn *v1alpha1.DNSet, sts *kruise.StatefulSet, sp v1alpha1.Shared common.SetupMemoryFsVolume(specRef, dn.Spec.MemoryFsSize) } -// buildDNSetConfigMap return dn set configmap -func buildDNSetConfigMap(dn *v1alpha1.DNSet, ls *v1alpha1.LogSet) (*corev1.ConfigMap, string, error) { +// buildDNSetConfigMap return dn set configmap. +// reservedOrdinals should be set to the LogSet StatefulSet's spec.reserveOrdinals so that +// service-addresses correctly skips ordinal holes created during failover (issue #596). +func buildDNSetConfigMap(dn *v1alpha1.DNSet, ls *v1alpha1.LogSet, reservedOrdinals []int) (*corev1.ConfigMap, string, error) { if ls.Status.Discovery == nil { return nil, "", errors.New("HAKeeper discovery address not ready") } @@ -217,7 +219,7 @@ func buildDNSetConfigMap(dn *v1alpha1.DNSet, ls *v1alpha1.LogSet) (*corev1.Confi // via discovery-address, operator can take off unhealthy logstores without restart CN/TN conf.Set([]string{"hakeeper-client", "discovery-address"}, ls.Status.Discovery.String()) } else { - conf.Set([]string{"hakeeper-client", "service-addresses"}, logset.HaKeeperAdds(ls)) + conf.Set([]string{"hakeeper-client", "service-addresses"}, logset.HaKeeperSvcAddrs(ls, reservedOrdinals)) } conf.MergeDeep(common.FileServiceConfig(fmt.Sprintf("%s/%s", common.DataPath, common.DataDir), ls.Spec.SharedStorage, &dn.Spec.SharedStorageCache)) conf.Set([]string{"service-type"}, serviceType) @@ -281,8 +283,8 @@ func syncPersistentVolumeClaim(dn *v1alpha1.DNSet, sts *kruise.StatefulSet) { } } -func syncPods(ctx *recon.Context[*v1alpha1.DNSet], sts *kruise.StatefulSet) error { - cm, configSuffix, err := buildDNSetConfigMap(ctx.Obj, ctx.Dep.Deps.LogSet) +func syncPods(ctx *recon.Context[*v1alpha1.DNSet], sts *kruise.StatefulSet, reservedOrdinals []int) error { + cm, configSuffix, err := buildDNSetConfigMap(ctx.Obj, ctx.Dep.Deps.LogSet, reservedOrdinals) if err != nil { return err } diff --git a/pkg/controllers/dnset/resource_test.go b/pkg/controllers/dnset/resource_test.go index 69b02720..f567eeaa 100644 --- a/pkg/controllers/dnset/resource_test.go +++ b/pkg/controllers/dnset/resource_test.go @@ -25,8 +25,9 @@ import ( func Test_buildDNSetConfigMap(t *testing.T) { type args struct { - dn *v1alpha1.DNSet - ls *v1alpha1.LogSet + dn *v1alpha1.DNSet + ls *v1alpha1.LogSet + reservedOrdinals []int } tests := []struct { name string @@ -239,13 +240,82 @@ memory-capacity = "1B" [hakeeper-client] discovery-address = "test:6001" +`, + }, + { + // regression test for #596: reservedOrdinals must be threaded through to + // HaKeeperSvcAddrs so that service-addresses skips the failed ordinal and + // includes the newly created replacement pod. + name: "1.x failover skips reserved ordinal", + args: args{ + dn: &v1alpha1.DNSet{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "test", + Name: "test", + }, + }, + ls: &v1alpha1.LogSet{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "test", + Name: "test", + }, + Spec: v1alpha1.LogSetSpec{ + PodSet: v1alpha1.PodSet{Replicas: 3}, + SharedStorage: v1alpha1.SharedStorageProvider{ + FileSystem: &v1alpha1.FileSystemProvider{ + Path: "/test", + }, + }, + }, + Status: v1alpha1.LogSetStatus{ + Discovery: &v1alpha1.LogSetDiscovery{ + Port: 6001, + Address: "test", + }, + }, + }, + reservedOrdinals: []int{1}, + }, + wantConfig: `data-dir = "/var/lib/matrixone/data" +service-type = "DN" + +[dn] +listen-address = "0.0.0.0:41010" +port-base = 41010 + +[dn.LogtailServer] +listen-address = "0.0.0.0:32003" + +[dn.lockservice] +listen-address = "0.0.0.0:6003" + +[[fileservice]] +backend = "DISK" +data-dir = "/var/lib/matrixone/data" +name = "LOCAL" + +[[fileservice]] +backend = "DISK" +data-dir = "/test" +name = "S3" + +[[fileservice]] +backend = "DISK-ETL" +data-dir = "/test" +name = "ETL" + +[fileservice.cache] +memory-capacity = "1B" + +[hakeeper-client] +service-addresses = ["test-log-0.test-log-headless.test.svc:32001", "test-log-2.test-log-headless.test.svc:32001", "test-log-3.test-log-headless.test.svc:32001"] `, }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { g := NewGomegaWithT(t) - got, configSuffix, err := buildDNSetConfigMap(tt.args.dn, tt.args.ls) + got, configSuffix, err := buildDNSetConfigMap(tt.args.dn, tt.args.ls, tt.args.reservedOrdinals) if (err != nil) != tt.wantErr { t.Errorf("buildDNSetConfigMap() error = %v, wantErr %v", err, tt.wantErr) return diff --git a/pkg/controllers/logset/configmap.go b/pkg/controllers/logset/configmap.go index c52a8dbd..03c62d9e 100644 --- a/pkg/controllers/logset/configmap.go +++ b/pkg/controllers/logset/configmap.go @@ -287,12 +287,22 @@ func buildConfigMap(ls *v1alpha1.LogSet) (*corev1.ConfigMap, string, error) { return cm, configSuffix, nil } -func HaKeeperAdds(ls *v1alpha1.LogSet) []string { - // TODO: consider hole in asts ordinals +// HaKeeperSvcAddrs returns the HAKeeper service addresses for the currently running LogStore pods, +// correctly skipping ordinal holes specified by reservedOrdinals (mirrors +// StatefulSet.spec.reserveOrdinals set during failover). +// Use this instead of HaKeeperAdds when the caller can supply the reserved-ordinal list from the +// underlying kruise StatefulSet (sts.Spec.ReserveOrdinals). +func HaKeeperSvcAddrs(ls *v1alpha1.LogSet, reservedOrdinals []int) []string { var seeds []string - for i := int32(0); i < ls.Spec.Replicas; i++ { + r := ls.Spec.Replicas + i := 0 + for count := int32(0); count < r; i++ { + if slices.Contains(reservedOrdinals, i) { + continue + } podName := fmt.Sprintf("%s-%d", stsName(ls), i) seeds = append(seeds, fmt.Sprintf("%s.%s.%s.svc:%d", podName, headlessSvcName(ls), ls.Namespace, logServicePort)) + count++ } return seeds } diff --git a/pkg/controllers/logset/configmap_test.go b/pkg/controllers/logset/configmap_test.go index 3be8628e..3ee0c784 100644 --- a/pkg/controllers/logset/configmap_test.go +++ b/pkg/controllers/logset/configmap_test.go @@ -80,3 +80,67 @@ func Test_gossipSeeds(t *testing.T) { }) } } + +func Test_HaKeeperSvcAddrs(t *testing.T) { + ls := &v1alpha1.LogSet{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: "default", + Name: "test", + }, + Spec: v1alpha1.LogSetSpec{ + PodSet: v1alpha1.PodSet{Replicas: 3}, + }, + } + tests := []struct { + name string + reservedOrdinals []int + want []string + }{ + { + name: "basic, no failover", + reservedOrdinals: nil, + want: []string{ + "test-log-0.test-log-headless.default.svc:32001", + "test-log-1.test-log-headless.default.svc:32001", + "test-log-2.test-log-headless.default.svc:32001", + }, + }, + { + // regression test for #596: HaKeeperAdds() previously ignored + // ReserveOrdinals entirely and always returned [log-0, log-1, log-2], + // pointing at the dead log-1 and missing the newly created log-3. + name: "log-1 failover creates log-3, hole must be skipped", + reservedOrdinals: []int{1}, + want: []string{ + "test-log-0.test-log-headless.default.svc:32001", + "test-log-2.test-log-headless.default.svc:32001", + "test-log-3.test-log-headless.default.svc:32001", + }, + }, + { + name: "log-0 failover creates log-3, hole must be skipped", + reservedOrdinals: []int{0}, + want: []string{ + "test-log-1.test-log-headless.default.svc:32001", + "test-log-2.test-log-headless.default.svc:32001", + "test-log-3.test-log-headless.default.svc:32001", + }, + }, + { + name: "reservation outside current window is a no-op", + reservedOrdinals: []int{3}, + want: []string{ + "test-log-0.test-log-headless.default.svc:32001", + "test-log-1.test-log-headless.default.svc:32001", + "test-log-2.test-log-headless.default.svc:32001", + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := HaKeeperSvcAddrs(ls, tt.reservedOrdinals); !reflect.DeepEqual(got, tt.want) { + t.Errorf("HaKeeperSvcAddrs() = %v, want %v", got, tt.want) + } + }) + } +} diff --git a/pkg/controllers/logset/sts.go b/pkg/controllers/logset/sts.go index f50d92a3..ebc94209 100644 --- a/pkg/controllers/logset/sts.go +++ b/pkg/controllers/logset/sts.go @@ -195,6 +195,13 @@ func stsName(ls *v1alpha1.LogSet) string { return resourceName(ls) } +// LogSetStsName returns the name of the kruise StatefulSet managed for the given LogSet. +// Exported so that other controllers (e.g. dnset, cnset) can look up the STS to read +// ReserveOrdinals without requiring a CRD change. +func LogSetStsName(ls *v1alpha1.LogSet) string { + return stsName(ls) +} + func headlessSvcName(ls *v1alpha1.LogSet) string { return resourceName(ls) + "-headless" }