milvus/pkg/taskcommon/state.go
wei liu 975c91df16
feat: Add comprehensive snapshot functionality for collections (#44361)
issue: #44358

Implement complete snapshot management system including creation,
deletion, listing, description, and restoration capabilities across all
system components.

Key features:
- Create snapshots for entire collections
- Drop snapshots by name with proper cleanup
- List snapshots with collection filtering
- Describe snapshot details and metadata

Components added/modified:
- Client SDK with full snapshot API support and options
- DataCoord snapshot service with metadata management
- Proxy layer with task-based snapshot operations
- Protocol buffer definitions for snapshot RPCs
- Comprehensive unit tests with mockey framework
- Integration tests for end-to-end validation

Technical implementation:
- Snapshot metadata storage in etcd with proper indexing
- File-based snapshot data persistence in object storage
- Garbage collection integration for snapshot cleanup
- Error handling and validation across all operations
- Thread-safe operations with proper locking mechanisms

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
- Core invariant/assumption: snapshots are immutable point‑in‑time
captures identified by (collection, snapshot name/ID); etcd snapshot
metadata is authoritative for lifecycle (PENDING → COMMITTED → DELETING)
and per‑segment manifests live in object storage (Avro / StorageV2). GC
and restore logic must see snapshotRefIndex loaded
(snapshotMeta.IsRefIndexLoaded) before reclaiming or relying on
segment/index files.

- New capability added: full end‑to‑end snapshot subsystem — client SDK
APIs (Create/Drop/List/Describe/Restore + restore job queries),
DataCoord SnapshotWriter/Reader (Avro + StorageV2 manifests),
snapshotMeta in meta, SnapshotManager orchestration
(create/drop/describe/list/restore), copy‑segment restore
tasks/inspector/checker, proxy & RPC surface, GC integration, and
docs/tests — enabling point‑in‑time collection snapshots persisted to
object storage and restorations orchestrated across components.

- Logic removed/simplified and why: duplicated recursive
compaction/delta‑log traversal and ad‑hoc lookup code were consolidated
behind two focused APIs/owners (Handler.GetDeltaLogFromCompactTo for
delta traversal and SnapshotManager/SnapshotReader for snapshot I/O).
MixCoord/coordinator broker paths were converted to thin RPC proxies.
This eliminates multiple implementations of the same traversal/lookup,
reducing divergence and simplifying responsibility boundaries.

- Why this does NOT introduce data loss or regressions: snapshot
create/drop use explicit two‑phase semantics (PENDING → COMMIT/DELETING)
with SnapshotWriter writing manifests and metadata before commit; GC
uses snapshotRefIndex guards and
IsRefIndexLoaded/GetSnapshotBySegment/GetSnapshotByIndex checks to avoid
removing referenced files; restore flow pre‑allocates job IDs, validates
resources (partitions/indexes), performs rollback on failure
(rollbackRestoreSnapshot), and converts/updates segment/index metadata
only after successful copy tasks. Extensive unit and integration tests
exercise pending/deleting/GC/restore/error paths to ensure idempotence
and protection against premature deletion.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Signed-off-by: Wei Liu <wei.liu@zilliz.com>
2026-01-06 10:15:24 +08:00

130 lines
3.9 KiB
Go

// Licensed to the LF AI & Data foundation under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you 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 taskcommon
import (
"github.com/milvus-io/milvus/pkg/v2/proto/datapb"
"github.com/milvus-io/milvus/pkg/v2/proto/indexpb"
)
type State = indexpb.JobState
const (
None = indexpb.JobState_JobStateNone
Init = indexpb.JobState_JobStateInit
InProgress = indexpb.JobState_JobStateInProgress
Finished = indexpb.JobState_JobStateFinished
Failed = indexpb.JobState_JobStateFailed
Retry = indexpb.JobState_JobStateRetry
)
func FromImportState(s datapb.ImportTaskStateV2) State {
switch s {
case datapb.ImportTaskStateV2_Pending:
return Init
case datapb.ImportTaskStateV2_InProgress:
return InProgress
case datapb.ImportTaskStateV2_Failed:
return Failed
case datapb.ImportTaskStateV2_Completed:
return Finished
case datapb.ImportTaskStateV2_Retry:
return Retry
}
return None
}
func ToImportState(s State) datapb.ImportTaskStateV2 {
switch s {
case Init:
return datapb.ImportTaskStateV2_Pending
case InProgress:
return datapb.ImportTaskStateV2_InProgress
case Failed:
return datapb.ImportTaskStateV2_Failed
case Finished:
return datapb.ImportTaskStateV2_Completed
case Retry:
return datapb.ImportTaskStateV2_Retry
}
return datapb.ImportTaskStateV2_None
}
func FromCompactionState(s datapb.CompactionTaskState) State {
switch s {
case datapb.CompactionTaskState_pipelining:
return Init
case datapb.CompactionTaskState_executing:
return InProgress
case datapb.CompactionTaskState_completed, datapb.CompactionTaskState_meta_saved,
datapb.CompactionTaskState_statistic, datapb.CompactionTaskState_indexing, datapb.CompactionTaskState_cleaned:
return Finished
case datapb.CompactionTaskState_failed:
return Failed
case datapb.CompactionTaskState_timeout:
return Retry
}
return None
}
func ToCompactionState(s State) datapb.CompactionTaskState {
switch s {
case Init:
return datapb.CompactionTaskState_pipelining
case InProgress:
return datapb.CompactionTaskState_executing
case Finished:
return datapb.CompactionTaskState_completed
case Failed:
return datapb.CompactionTaskState_failed
case Retry:
return datapb.CompactionTaskState_timeout
}
return datapb.CompactionTaskState_unknown
}
func FromCopySegmentState(s datapb.CopySegmentTaskState) State {
switch s {
case datapb.CopySegmentTaskState_CopySegmentTaskPending:
return Init
case datapb.CopySegmentTaskState_CopySegmentTaskInProgress:
return InProgress
case datapb.CopySegmentTaskState_CopySegmentTaskCompleted:
return Finished
case datapb.CopySegmentTaskState_CopySegmentTaskFailed:
return Failed
}
return None
}
func ToCopySegmentState(s State) datapb.CopySegmentTaskState {
switch s {
case Init:
return datapb.CopySegmentTaskState_CopySegmentTaskPending
case InProgress:
return datapb.CopySegmentTaskState_CopySegmentTaskInProgress
case Finished:
return datapb.CopySegmentTaskState_CopySegmentTaskCompleted
case Failed:
return datapb.CopySegmentTaskState_CopySegmentTaskFailed
case Retry:
// CopySegmentTask does not support Retry state, map to Failed
return datapb.CopySegmentTaskState_CopySegmentTaskFailed
}
return datapb.CopySegmentTaskState_CopySegmentTaskNone
}