From ee361cd05302c48afa55175127350778f4456034 Mon Sep 17 00:00:00 2001 From: lr90 Date: Tue, 1 Sep 2026 18:00:11 +0800 Subject: [PATCH] fix: fail closed when draining CN pipelines (#616) --- api/core/v1alpha1/semver.go | 2 +- api/core/v1alpha1/semver_test.go | 16 ++- pkg/controllers/cnstore/controller.go | 44 ++++++-- pkg/controllers/cnstore/stats_test.go | 125 +++++++++++++++++++++ pkg/controllers/common/cnstore.go | 31 ++++-- pkg/controllers/common/cnstore_test.go | 144 +++++++++++++++++++++++++ 6 files changed, 337 insertions(+), 25 deletions(-) create mode 100644 pkg/controllers/cnstore/stats_test.go diff --git a/api/core/v1alpha1/semver.go b/api/core/v1alpha1/semver.go index 43305f17..c896bebe 100644 --- a/api/core/v1alpha1/semver.go +++ b/api/core/v1alpha1/semver.go @@ -35,7 +35,7 @@ var ( // 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")}, + MOFeaturePipelineInfo: {semver.MustParse("1.1.2"), semver.MustParse("1.2.0"), semver.MustParse("2.0.0"), semver.MustParse("3.0.0"), semver.MustParse("4.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")}, diff --git a/api/core/v1alpha1/semver_test.go b/api/core/v1alpha1/semver_test.go index 854a1a4f..ed05e81e 100644 --- a/api/core/v1alpha1/semver_test.go +++ b/api/core/v1alpha1/semver_test.go @@ -41,6 +41,13 @@ func TestHasMOFeature(t *testing.T) { g.Expect(HasMOFeature(mustParse("1.1.3"), MOFeaturePipelineInfo)).To(BeTrue()) g.Expect(HasMOFeature(mustParse("1.2.0"), MOFeaturePipelineInfo)).To(BeTrue()) g.Expect(HasMOFeature(mustParse("2.0.0"), MOFeaturePipelineInfo)).To(BeTrue()) + g.Expect(HasMOFeature(mustParse("3.0.0"), MOFeaturePipelineInfo)).To(BeTrue()) + g.Expect(HasMOFeature(mustParse("3.9.9"), MOFeaturePipelineInfo)).To(BeTrue()) + g.Expect(HasMOFeature(mustParse("v3.0.10-90ae8dc0c-2026-04-22"), MOFeaturePipelineInfo)).To(BeTrue()) + g.Expect(HasMOFeature(mustParse("4.0.0"), MOFeaturePipelineInfo)).To(BeTrue()) + g.Expect(HasMOFeature(mustParse("v4.0.0-rc1"), MOFeaturePipelineInfo)).To(BeTrue()) + g.Expect(HasMOFeature(mustParse("4.2.1"), MOFeaturePipelineInfo)).To(BeTrue()) + g.Expect(HasMOFeature(mustParse("v4.2.1-d2393868a-2026-08-28"), MOFeaturePipelineInfo)).To(BeTrue()) g.Expect(HasMOFeature(mustParse("1.1.1"), MOFeaturePipelineInfo)).To(BeFalse()) g.Expect(HasMOFeature(mustParse("1.1.1"), MOFeaturePipelineInfo)).To(BeFalse()) g.Expect(HasMOFeature(mustParse("1.1.2"), MOFeaturePipelineInfo)).To(BeTrue()) @@ -79,14 +86,13 @@ func TestHasMOFeature_DiscoveryFixed(t *testing.T) { // 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. +// unrelated behavior (lock migration handshake, sharding stats collection, session source +// accounting) on MO 3.x without dedicated verification. DiscoveryFixed and PipelineInfo have +// been verified independently; this test should be updated deliberately, one feature at a time, +// as each remaining feature is verified. func TestHasMOFeature_OtherFeaturesNotExtendedTo3x(t *testing.T) { g := NewGomegaWithT(t) unverifiedOn3x := []MOFeature{ - MOFeaturePipelineInfo, MOFeatureSessionSource, MOFeatureLockMigration, MOFeatureShardingMigration, diff --git a/pkg/controllers/cnstore/controller.go b/pkg/controllers/cnstore/controller.go index 6d81b5ed..9cea25d4 100644 --- a/pkg/controllers/cnstore/controller.go +++ b/pkg/controllers/cnstore/controller.go @@ -34,6 +34,7 @@ import ( "github.com/matrixorigin/matrixone-operator/pkg/querycli" logpb "github.com/matrixorigin/matrixone/pkg/pb/logservice" "github.com/matrixorigin/matrixone/pkg/pb/metadata" + querypb "github.com/matrixorigin/matrixone/pkg/pb/query" "github.com/openkruise/kruise-api/apps/pub" kruisev1alpha1 "github.com/openkruise/kruise-api/apps/v1alpha1" corev1 "k8s.io/api/core/v1" @@ -69,7 +70,13 @@ const resyncInterval = 30 * time.Second type Controller struct { clientMgr *mocli.MORPCClientManager - queryCli *querycli.Client + queryCli queryClient +} + +type queryClient interface { + ShowProcessList(context.Context, string) (*querypb.ShowProcessListResponse, error) + GetPipelineInfo(context.Context, string) (*querypb.GetPipelineInfoResponse, error) + GetReplicaCount(context.Context, string) (querypb.GetReplicaCountResponse, error) } type withCNSet struct { @@ -492,10 +499,9 @@ func (c *withCNSet) syncStats(ctx *recon.Context[*corev1.Pod]) error { if err == nil { sc = previous } - if sc.StartedTime == nil || !sc.StartedTime.Equal(*startedTime) { - // clean previously recorded score and update startTime if CN is restarted - sc.Restarted(startedTime) - } + // Invalidate the previous round before performing any query. A failed or partial + // refresh must never leave an old zero looking like current evidence. + sc.BeginObservation(startedTime) uid := v1alpha1.GetCNPodUUID(pod) moVersion := common.GetSemanticVersion(&pod.ObjectMeta) @@ -509,6 +515,9 @@ func (c *withCNSet) syncStats(ctx *recon.Context[*corev1.Pod]) error { return nil }); err != nil { ctx.Log.Info("error refresh stats, cn not found in store-cache", "error", err.Error()) + // BeginObservation has already invalidated this round. Persist that state so + // IsSafeToReclaim remains the single reclaim-safety boundary, while preserving + // the existing state-sync behavior when the CN is absent from the cache. return c.patchStoreStats(ctx, sc) } @@ -517,39 +526,52 @@ func (c *withCNSet) syncStats(ctx *recon.Context[*corev1.Pod]) error { Logger: ctx.Log, Enabled: diagnosDraining, } + c.collectQueryStats(sc, queryAddress, moVersion, diagosis) + + return c.patchStoreStats(ctx, sc) +} + +// collectQueryStats updates one observation round. Each observation is marked +// valid only after its query succeeds. A feature that is unavailable for the +// current MO version is explicitly treated as not required with a zero count. +func (c *Controller) collectQueryStats(sc *common.StoreScore, queryAddress string, moVersion semver.Version, diagosis *connectionDiagnosis) { count, err := c.getSessionCount(queryAddress, moVersion, diagosis) if err != nil { - ctx.Log.Info("error get session count", "error", err.Error()) + diagosis.Logger.Info("error get session count", "error", err.Error()) } else { // update session count sc.SessionCount = count + sc.SessionObserved = true } var pipelineCount int if v1alpha1.HasMOFeature(moVersion, v1alpha1.MOFeaturePipelineInfo) { pipelineCount, err = c.getPipelineCount(queryAddress, diagosis) if err != nil { - ctx.Log.Info("error get pipeline count", "error", err.Error()) + diagosis.Logger.Info("error get pipeline count", "error", err.Error()) } else { // update pipeline count sc.PipelineCount = pipelineCount + sc.PipelineObserved = true } } else { - // clear pipeline count if feature is disabled + // Pipeline observation is not required for versions without the API. sc.PipelineCount = 0 + sc.PipelineObserved = true } var replicaCount int if v1alpha1.HasMOFeature(moVersion, v1alpha1.MOFeatureShardingMigration) { replicaCount, err = c.getReplicaCount(queryAddress, diagosis) if err != nil { - ctx.Log.Info("error get replica count", "error", err.Error()) + diagosis.Logger.Info("error get replica count", "error", err.Error()) } else { sc.ReplicaCount = replicaCount + sc.ReplicaObserved = true } } else { + // Replica observation is not required for versions without sharding migration. sc.ReplicaCount = 0 + sc.ReplicaObserved = true } - - return c.patchStoreStats(ctx, sc) } func (c *Controller) patchStoreStats(ctx *recon.Context[*corev1.Pod], sc *common.StoreScore) error { diff --git a/pkg/controllers/cnstore/stats_test.go b/pkg/controllers/cnstore/stats_test.go new file mode 100644 index 00000000..abc6eb46 --- /dev/null +++ b/pkg/controllers/cnstore/stats_test.go @@ -0,0 +1,125 @@ +// Copyright 2026 Matrix Origin +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package cnstore + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/blang/semver/v4" + "github.com/go-logr/logr" + "github.com/matrixorigin/matrixone-operator/pkg/controllers/common" + querypb "github.com/matrixorigin/matrixone/pkg/pb/query" +) + +type fakeQueryClient struct { + pipelineCount int64 + replicaCount int64 + sessionErr error + pipelineErr error + replicaErr error + + pipelineCalls int + replicaCalls int +} + +func (f *fakeQueryClient) ShowProcessList(context.Context, string) (*querypb.ShowProcessListResponse, error) { + if f.sessionErr != nil { + return nil, f.sessionErr + } + return &querypb.ShowProcessListResponse{}, nil +} + +func (f *fakeQueryClient) GetPipelineInfo(context.Context, string) (*querypb.GetPipelineInfoResponse, error) { + f.pipelineCalls++ + if f.pipelineErr != nil { + return nil, f.pipelineErr + } + return &querypb.GetPipelineInfoResponse{Count: f.pipelineCount}, nil +} + +func (f *fakeQueryClient) GetReplicaCount(context.Context, string) (querypb.GetReplicaCountResponse, error) { + f.replicaCalls++ + if f.replicaErr != nil { + return querypb.GetReplicaCountResponse{}, f.replicaErr + } + return querypb.GetReplicaCountResponse{Count: f.replicaCount}, nil +} + +func TestCollectQueryStatsFailClosed(t *testing.T) { + tests := []struct { + name string + version string + query *fakeQueryClient + safe bool + pipelineCalls int + replicaCalls int + }{ + { + name: "MO 4 complete zero", + version: "4.2.0", + query: &fakeQueryClient{}, + safe: true, + pipelineCalls: 1, + }, + { + name: "MO 4 active AP pipeline", + version: "4.2.0", + query: &fakeQueryClient{pipelineCount: 1}, + pipelineCalls: 1, + }, + { + name: "MO 4 pipeline query failure", + version: "4.2.0", + query: &fakeQueryClient{pipelineErr: errors.New("pipeline unavailable")}, + pipelineCalls: 1, + }, + { + name: "session query failure", + version: "4.2.0", + query: &fakeQueryClient{sessionErr: errors.New("session unavailable")}, + pipelineCalls: 1, + }, + { + name: "required replica query failure", + version: "2.0.0", + query: &fakeQueryClient{replicaErr: errors.New("replica unavailable")}, + pipelineCalls: 1, + replicaCalls: 1, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + started := time.Now() + score := &common.StoreScore{StartedTime: &started} + score.BeginObservation(&started) + controller := &Controller{queryCli: tt.query} + controller.collectQueryStats(score, "query-address", semver.MustParse(tt.version), &connectionDiagnosis{Logger: logr.Discard()}) + + if got := score.IsSafeToReclaim(); got != tt.safe { + t.Fatalf("IsSafeToReclaim() = %v, want %v; score=%+v", got, tt.safe, score) + } + if tt.query.pipelineCalls != tt.pipelineCalls { + t.Fatalf("pipeline calls = %d, want %d", tt.query.pipelineCalls, tt.pipelineCalls) + } + if tt.query.replicaCalls != tt.replicaCalls { + t.Fatalf("replica calls = %d, want %d", tt.query.replicaCalls, tt.replicaCalls) + } + }) + } +} diff --git a/pkg/controllers/common/cnstore.go b/pkg/controllers/common/cnstore.go index 848a02de..a5f07a7c 100644 --- a/pkg/controllers/common/cnstore.go +++ b/pkg/controllers/common/cnstore.go @@ -151,6 +151,12 @@ type StoreScore struct { SessionCount int `json:"sessionCount"` PipelineCount int `json:"pipelineCount"` ReplicaCount int `json:"replicaCount"` + // The observed fields distinguish a confirmed count from a zero value left by a + // skipped, failed, or incomplete query. Old annotations do not contain these + // fields and are deliberately unsafe until a fresh observation succeeds. + SessionObserved bool `json:"sessionObserved"` + PipelineObserved bool `json:"pipelineObserved"` + ReplicaObserved bool `json:"replicaObserved"` StartedTime *time.Time `json:"startedTime,omitempty"` } @@ -160,14 +166,23 @@ func (s *StoreScore) GenDeletionCost() int { } func (s *StoreScore) IsSafeToReclaim() bool { - return s.SessionCount == 0 && s.PipelineCount == 0 && s.ReplicaCount == 0 -} - -func (s *StoreScore) Restarted(startedTime *time.Time) { - s.SessionCount = 0 - s.PipelineCount = 0 - s.ReplicaCount = 0 - s.StartedTime = startedTime + return s.SessionObserved && s.PipelineObserved && s.ReplicaObserved && + s.SessionCount == 0 && s.PipelineCount == 0 && s.ReplicaCount == 0 +} + +// BeginObservation invalidates the previous round before any query starts. Counts +// are kept for diagnostics and deletion cost unless the CN process restarted, but +// they cannot authorize reclaim until every required observation succeeds again. +func (s *StoreScore) BeginObservation(startedTime *time.Time) { + s.SessionObserved = false + s.PipelineObserved = false + s.ReplicaObserved = false + if s.StartedTime == nil || !s.StartedTime.Equal(*startedTime) { + s.SessionCount = 0 + s.PipelineCount = 0 + s.ReplicaCount = 0 + s.StartedTime = startedTime + } } // GetStoreScore get the store connection count from Pod anno diff --git a/pkg/controllers/common/cnstore_test.go b/pkg/controllers/common/cnstore_test.go index 14e38ec5..1721b246 100644 --- a/pkg/controllers/common/cnstore_test.go +++ b/pkg/controllers/common/cnstore_test.go @@ -16,10 +16,154 @@ package common import ( "testing" + "time" + "github.com/matrixorigin/matrixone-operator/api/core/v1alpha1" corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) +func TestStoreScoreIsSafeToReclaim(t *testing.T) { + now := time.Now() + tests := []struct { + name string + score StoreScore + safe bool + }{ + { + name: "complete zero observation", + score: StoreScore{ + SessionObserved: true, PipelineObserved: true, ReplicaObserved: true, + StartedTime: &now, + }, + safe: true, + }, + { + name: "active pipeline without ordinary sessions", + score: StoreScore{ + PipelineCount: 1, + SessionObserved: true, PipelineObserved: true, ReplicaObserved: true, + StartedTime: &now, + }, + }, + { + name: "active session", + score: StoreScore{ + SessionCount: 1, + SessionObserved: true, PipelineObserved: true, ReplicaObserved: true, + StartedTime: &now, + }, + }, + { + name: "active replica", + score: StoreScore{ + ReplicaCount: 1, + SessionObserved: true, PipelineObserved: true, ReplicaObserved: true, + StartedTime: &now, + }, + }, + { + name: "legacy zero without observation evidence", + score: StoreScore{StartedTime: &now}, + }, + { + name: "pipeline query failed", + score: StoreScore{ + SessionObserved: true, ReplicaObserved: true, + StartedTime: &now, + }, + }, + { + name: "session query failed", + score: StoreScore{ + PipelineObserved: true, ReplicaObserved: true, + StartedTime: &now, + }, + }, + { + name: "replica query failed", + score: StoreScore{ + SessionObserved: true, PipelineObserved: true, + StartedTime: &now, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := tt.score.IsSafeToReclaim(); got != tt.safe { + t.Fatalf("IsSafeToReclaim() = %v, want %v", got, tt.safe) + } + }) + } +} + +func TestGetStoreScoreLegacyAnnotationsAreUnsafe(t *testing.T) { + tests := []struct { + name string + value string + }{ + {name: "legacy JSON zero", value: `{"sessionCount":0,"pipelineCount":0,"replicaCount":0}`}, + {name: "legacy integer zero", value: "0"}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + pod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Annotations: map[string]string{v1alpha1.StoreScoreAnno: tt.value}}} + score, err := GetStoreScore(pod) + if err != nil { + t.Fatalf("GetStoreScore() error = %v", err) + } + if score.IsSafeToReclaim() { + t.Fatal("legacy score without observation evidence must not authorize reclaim") + } + }) + } +} + +func TestStoreScoreBeginObservationInvalidatesStaleZero(t *testing.T) { + started := time.Now() + score := StoreScore{ + SessionObserved: true, + PipelineObserved: true, + ReplicaObserved: true, + StartedTime: &started, + } + if !score.IsSafeToReclaim() { + t.Fatal("precondition: complete zero observation should be safe") + } + + score.BeginObservation(&started) + if score.IsSafeToReclaim() { + t.Fatal("a prior zero must not authorize reclaim in a new observation round") + } +} + +func TestStoreScoreBeginObservationResetsRestartedCN(t *testing.T) { + previousStart := time.Now().Add(-time.Minute) + currentStart := time.Now() + score := StoreScore{ + SessionCount: 3, + PipelineCount: 2, + ReplicaCount: 1, + SessionObserved: true, + PipelineObserved: true, + ReplicaObserved: true, + StartedTime: &previousStart, + } + + score.BeginObservation(¤tStart) + if score.SessionCount != 0 || score.PipelineCount != 0 || score.ReplicaCount != 0 { + t.Fatalf("restart did not reset counts: %+v", score) + } + if score.SessionObserved || score.PipelineObserved || score.ReplicaObserved { + t.Fatalf("restart retained observation validity: %+v", score) + } + if score.StartedTime == nil || !score.StartedTime.Equal(currentStart) { + t.Fatalf("StartedTime = %v, want %v", score.StartedTime, currentStart) + } +} + // returns true when pod spec container image differs from pod status container image func TestNeedUpdateImage_DifferentImages(t *testing.T) { pod := &corev1.Pod{