forked from mooncake-track/Mooncake
Compare commits
4 Commits
main
...
copilot/mo
| Author | SHA1 | Date |
|---|---|---|
 zhang zuoyuan
|
9408d853d6 | |
|
zhang zuoyuan
|
23b5886e8c | |
copilot-swe-agent[bot]
|
67b3f9de2e | |
|
copilot-swe-agent[bot]
|
ee69b7ea40 |
|
|
@ -5,7 +5,6 @@ build_ofed4
|
|||
old
|
||||
local_test
|
||||
go.sum
|
||||
!mooncake-common/etcd/go.sum
|
||||
*.so
|
||||
bin
|
||||
mod
|
||||
|
|
@ -205,4 +204,4 @@ CLAUDE.md
|
|||
_codeql_detected_source_root
|
||||
|
||||
# CodeBuddy Memory
|
||||
.codebuddy/
|
||||
.codebuddy/
|
||||
|
|
@ -45,16 +45,6 @@ option(STORE_USE_REDIS "build mooncake store with redis" OFF)
|
|||
if (STORE_USE_REDIS)
|
||||
add_compile_definitions(STORE_USE_REDIS)
|
||||
endif()
|
||||
option(STORE_USE_K8S_LEASE "build mooncake store with K8s Lease leader election" OFF)
|
||||
if (STORE_USE_K8S_LEASE)
|
||||
if (STORE_USE_ETCD)
|
||||
message(FATAL_ERROR "STORE_USE_K8S_LEASE and STORE_USE_ETCD cannot be enabled together because both build Go c-shared HA backends.")
|
||||
endif()
|
||||
if (USE_ETCD AND NOT USE_ETCD_LEGACY)
|
||||
message(FATAL_ERROR "STORE_USE_K8S_LEASE cannot be enabled with non-legacy USE_ETCD because both build Go c-shared libraries in the same process.")
|
||||
endif()
|
||||
add_compile_definitions(STORE_USE_K8S_LEASE)
|
||||
endif()
|
||||
|
||||
option(STORE_USE_JEMALLOC "Use jemalloc in mooncake store master" OFF)
|
||||
|
||||
|
|
|
|||
|
|
@ -23,7 +23,7 @@ NC="\033[0m" # No Color
|
|||
# Configuration
|
||||
REPO_ROOT=`pwd`
|
||||
GITHUB_PROXY=${GITHUB_PROXY:-"https://github.com"}
|
||||
GOVER=1.25.9
|
||||
GOVER=1.23.8
|
||||
|
||||
# Function to print section headers
|
||||
print_section() {
|
||||
|
|
|
|||
|
|
@ -2,10 +2,6 @@ if ((USE_ETCD AND NOT USE_ETCD_LEGACY) OR STORE_USE_ETCD)
|
|||
add_subdirectory(etcd)
|
||||
endif()
|
||||
|
||||
if (STORE_USE_K8S_LEASE)
|
||||
add_subdirectory(k8s-lease)
|
||||
endif()
|
||||
|
||||
include_directories(${CMAKE_CURRENT_SOURCE_DIR}/include)
|
||||
add_subdirectory(src)
|
||||
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ add_custom_command(
|
|||
COMMAND bash -c "go mod tidy" && bash -c "go build -buildmode=c-shared -o ${CMAKE_CURRENT_BINARY_DIR}/libetcd_wrapper.so etcd_wrapper.go" && cp ${CMAKE_CURRENT_BINARY_DIR}/libetcd_wrapper.h ${CMAKE_CURRENT_SOURCE_DIR}
|
||||
WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}
|
||||
COMMENT "Building Go shared library"
|
||||
DEPENDS etcd_wrapper.go go.mod go.sum build.sh
|
||||
DEPENDS etcd_wrapper.go
|
||||
)
|
||||
|
||||
set(ETCD_WRAPPER_INCLUDE ${CMAKE_CURRENT_BINARY_DIR}/libetcd_wrapper.h)
|
||||
|
|
@ -17,4 +17,4 @@ add_custom_target(
|
|||
install(
|
||||
FILES ${ETCD_WRAPPER_LIB}
|
||||
DESTINATION lib
|
||||
)
|
||||
)
|
||||
|
|
@ -1,8 +1,6 @@
|
|||
module github.com/kvcache-ai/Mooncake/mooncake-common/etcd
|
||||
|
||||
go 1.25.0
|
||||
|
||||
toolchain go1.25.9
|
||||
go 1.24.0
|
||||
|
||||
require (
|
||||
go.etcd.io/etcd/api/v3 v3.5.21
|
||||
|
|
|
|||
|
|
@ -1,108 +0,0 @@
|
|||
github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
|
||||
github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/coreos/go-semver v0.3.0 h1:wkHLiw0WNATZnSG7epLsujiMCgPAc9xhjJ4tgnAxmfM=
|
||||
github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk=
|
||||
github.com/coreos/go-systemd/v22 v22.3.2 h1:D9/bQk5vlXQFZ6Kwuu6zaiXJ9oTPe68++AzAJc1DzSI=
|
||||
github.com/coreos/go-systemd/v22 v22.3.2/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
|
||||
github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
|
||||
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
|
||||
github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
|
||||
github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
|
||||
github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
|
||||
github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
|
||||
github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek=
|
||||
github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps=
|
||||
github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
|
||||
github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
|
||||
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
|
||||
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
|
||||
github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
|
||||
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
|
||||
github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
|
||||
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
|
||||
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
|
||||
github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
|
||||
github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
|
||||
github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
|
||||
go.etcd.io/etcd/api/v3 v3.5.21 h1:A6O2/JDb3tvHhiIz3xf9nJ7REHvtEFJJ3veW3FbCnS8=
|
||||
go.etcd.io/etcd/api/v3 v3.5.21/go.mod h1:c3aH5wcvXv/9dqIw2Y810LDXJfhSYdHQ0vxmP3CCHVY=
|
||||
go.etcd.io/etcd/client/pkg/v3 v3.5.21 h1:lPBu71Y7osQmzlflM9OfeIV2JlmpBjqBNlLtcoBqUTc=
|
||||
go.etcd.io/etcd/client/pkg/v3 v3.5.21/go.mod h1:BgqT/IXPjK9NkeSDjbzwsHySX3yIle2+ndz28nVsjUs=
|
||||
go.etcd.io/etcd/client/v3 v3.5.21 h1:T6b1Ow6fNjOLOtM0xSoKNQt1ASPCLWrF9XMHcH9pEyY=
|
||||
go.etcd.io/etcd/client/v3 v3.5.21/go.mod h1:mFYy67IOqmbRf/kRUvsHixzo3iG+1OF2W2+jVIQRAnU=
|
||||
go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64=
|
||||
go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y=
|
||||
go.opentelemetry.io/otel v1.39.0 h1:8yPrr/S0ND9QEfTfdP9V+SiwT4E0G7Y5MO7p85nis48=
|
||||
go.opentelemetry.io/otel v1.39.0/go.mod h1:kLlFTywNWrFyEdH0oj2xK0bFYZtHRYUdv1NklR/tgc8=
|
||||
go.opentelemetry.io/otel/metric v1.39.0 h1:d1UzonvEZriVfpNKEVmHXbdf909uGTOQjA0HF0Ls5Q0=
|
||||
go.opentelemetry.io/otel/metric v1.39.0/go.mod h1:jrZSWL33sD7bBxg1xjrqyDjnuzTUB0x1nBERXd7Ftcs=
|
||||
go.opentelemetry.io/otel/sdk v1.39.0 h1:nMLYcjVsvdui1B/4FRkwjzoRVsMK8uL/cj0OyhKzt18=
|
||||
go.opentelemetry.io/otel/sdk v1.39.0/go.mod h1:vDojkC4/jsTJsE+kh+LXYQlbL8CgrEcwmt1ENZszdJE=
|
||||
go.opentelemetry.io/otel/sdk/metric v1.39.0 h1:cXMVVFVgsIf2YL6QkRF4Urbr/aMInf+2WKg+sEJTtB8=
|
||||
go.opentelemetry.io/otel/sdk/metric v1.39.0/go.mod h1:xq9HEVH7qeX69/JnwEfp6fVq5wosJsY1mt4lLfYdVew=
|
||||
go.opentelemetry.io/otel/trace v1.39.0 h1:2d2vfpEDmCJ5zVYz7ijaJdOF59xLomrvj7bjt6/qCJI=
|
||||
go.opentelemetry.io/otel/trace v1.39.0/go.mod h1:88w4/PnZSazkGzz/w84VHpQafiU4EtqqlVdxWy+rNOA=
|
||||
go.uber.org/atomic v1.7.0 h1:ADUqmZGgLDDfbSL9ZmPxKTybcoEYHgpYfELNoN+7hsw=
|
||||
go.uber.org/atomic v1.7.0/go.mod h1:fEN4uk6kAWBTFdckzkM89CLk9XfWZrxpCo0nPH17wJc=
|
||||
go.uber.org/multierr v1.6.0 h1:y6IPFStTAIT5Ytl7/XYmHvzXQ7S3g/IeZW9hyZ5thw4=
|
||||
go.uber.org/multierr v1.6.0/go.mod h1:cdWPpRnG4AhwMwsgIHip0KRBQjJy5kYEpYjJxpXp9iU=
|
||||
go.uber.org/zap v1.17.0 h1:MTjgFu6ZLKvY6Pvaqk97GlxNBuMpV4Hy/3P6tRGlI2U=
|
||||
go.uber.org/zap v1.17.0/go.mod h1:MXVU+bhUf/A7Xi2HNOnopQOrmycQ5Ih87HtOu4q5SSo=
|
||||
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
|
||||
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
|
||||
golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
|
||||
golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
|
||||
golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
|
||||
golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
|
||||
golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
|
||||
golang.org/x/net v0.48.0 h1:zyQRTTrjc33Lhh0fBgT/H3oZq9WuvRR5gPC70xpDiQU=
|
||||
golang.org/x/net v0.48.0/go.mod h1:+ndRgGjkh8FGtu1w1FGbEC31if4VrNVMuKTgcAAnQRY=
|
||||
golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.39.0 h1:CvCKL8MeisomCi6qNZ+wbb0DN9E5AATixKsvNtMoMFk=
|
||||
golang.org/x/sys v0.39.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
|
||||
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.32.0 h1:ZD01bjUt1FQ9WJ0ClOL5vxgxOI/sVCNgX1YtKwcY0mU=
|
||||
golang.org/x/text v0.32.0/go.mod h1:o/rUWzghvpD5TXrTIBuJU77MTaN0ljMWE47kxGJQ7jY=
|
||||
golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
|
||||
golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
|
||||
golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE=
|
||||
golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
|
||||
golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
|
||||
gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk=
|
||||
gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E=
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217 h1:fCvbg86sFXwdrl5LgVcTEvNC+2txB5mgROGmRL5mrls=
|
||||
google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217/go.mod h1:+rXWjjaukWZun3mLfjmVnQi18E1AsFbDN9QdJ5YXLto=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 h1:gRkg/vSppuSQoDjxyiGfN4Upv/h/DQmIR10ZU8dh4Ww=
|
||||
google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk=
|
||||
google.golang.org/grpc v1.79.3 h1:sybAEdRIEtvcD68Gx7dmnwjZKlyfuc61Dyo9pGXXkKE=
|
||||
google.golang.org/grpc v1.79.3/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ=
|
||||
google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE=
|
||||
google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
|
||||
gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
|
||||
gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
|
||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
|
||||
gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
|
|
@ -1,20 +0,0 @@
|
|||
add_custom_command(
|
||||
OUTPUT ${CMAKE_CURRENT_BINARY_DIR}/libk8s_lease_wrapper.so
|
||||
COMMAND bash -c "go mod tidy" && bash -c "go build -buildmode=c-shared -o ${CMAKE_CURRENT_BINARY_DIR}/libk8s_lease_wrapper.so k8s_lease_wrapper.go" && cp ${CMAKE_CURRENT_BINARY_DIR}/libk8s_lease_wrapper.h ${CMAKE_CURRENT_SOURCE_DIR}
|
||||
WORKING_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}
|
||||
COMMENT "Building K8s Lease Go shared library"
|
||||
DEPENDS k8s_lease_wrapper.go
|
||||
)
|
||||
|
||||
set(K8S_LEASE_WRAPPER_INCLUDE ${CMAKE_CURRENT_BINARY_DIR}/libk8s_lease_wrapper.h)
|
||||
set(K8S_LEASE_WRAPPER_LIB ${CMAKE_CURRENT_BINARY_DIR}/libk8s_lease_wrapper.so)
|
||||
|
||||
add_custom_target(
|
||||
build_k8s_lease_wrapper
|
||||
DEPENDS ${K8S_LEASE_WRAPPER_LIB}
|
||||
)
|
||||
|
||||
install(
|
||||
FILES ${K8S_LEASE_WRAPPER_LIB}
|
||||
DESTINATION lib
|
||||
)
|
||||
|
|
@ -1,61 +0,0 @@
|
|||
// envtest-server starts a real kube-apiserver + etcd via envtest, writes the
|
||||
// KUBECONFIG path to stdout, and blocks until SIGTERM or SIGINT. This lets
|
||||
// C++ tests launch it as a subprocess and talk to a real K8s API without a
|
||||
// full cluster.
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"os/signal"
|
||||
"path/filepath"
|
||||
"syscall"
|
||||
|
||||
"k8s.io/client-go/tools/clientcmd"
|
||||
clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
|
||||
"sigs.k8s.io/controller-runtime/pkg/envtest"
|
||||
)
|
||||
|
||||
func main() {
|
||||
env := &envtest.Environment{}
|
||||
|
||||
cfg, err := env.Start()
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "envtest start failed: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
// Write a KUBECONFIG file that points at the envtest kube-apiserver.
|
||||
kubeconfigPath := filepath.Join(os.TempDir(), fmt.Sprintf("envtest-kubeconfig-%d", os.Getpid()))
|
||||
kubeconfig := clientcmdapi.NewConfig()
|
||||
kubeconfig.Clusters["envtest"] = &clientcmdapi.Cluster{
|
||||
Server: cfg.Host,
|
||||
CertificateAuthorityData: cfg.CAData,
|
||||
}
|
||||
kubeconfig.AuthInfos["envtest"] = &clientcmdapi.AuthInfo{
|
||||
ClientCertificateData: cfg.CertData,
|
||||
ClientKeyData: cfg.KeyData,
|
||||
}
|
||||
kubeconfig.Contexts["envtest"] = &clientcmdapi.Context{
|
||||
Cluster: "envtest",
|
||||
AuthInfo: "envtest",
|
||||
}
|
||||
kubeconfig.CurrentContext = "envtest"
|
||||
|
||||
if err := clientcmd.WriteToFile(*kubeconfig, kubeconfigPath); err != nil {
|
||||
fmt.Fprintf(os.Stderr, "failed to write kubeconfig: %v\n", err)
|
||||
env.Stop()
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
// Print the kubeconfig path — the parent process reads this from stdout.
|
||||
fmt.Println(kubeconfigPath)
|
||||
|
||||
// Block until SIGTERM or SIGINT.
|
||||
sigCh := make(chan os.Signal, 1)
|
||||
signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT)
|
||||
<-sigCh
|
||||
|
||||
os.Remove(kubeconfigPath)
|
||||
env.Stop()
|
||||
}
|
||||
|
|
@ -1,60 +0,0 @@
|
|||
module github.com/kvcache-ai/Mooncake/mooncake-common/k8s-lease
|
||||
|
||||
go 1.24.0
|
||||
|
||||
require (
|
||||
k8s.io/api v0.34.3
|
||||
k8s.io/apimachinery v0.34.3
|
||||
k8s.io/client-go v0.34.3
|
||||
k8s.io/utils v0.0.0-20251002143259-bc988d571ff4
|
||||
sigs.k8s.io/controller-runtime v0.22.5
|
||||
)
|
||||
|
||||
require (
|
||||
github.com/beorn7/perks v1.0.1 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||
github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
|
||||
github.com/emicklei/go-restful/v3 v3.12.2 // indirect
|
||||
github.com/evanphx/json-patch/v5 v5.9.11 // indirect
|
||||
github.com/fxamacker/cbor/v2 v2.9.0 // indirect
|
||||
github.com/go-logr/logr v1.4.3 // indirect
|
||||
github.com/go-openapi/jsonpointer v0.21.0 // indirect
|
||||
github.com/go-openapi/jsonreference v0.20.2 // indirect
|
||||
github.com/go-openapi/swag v0.23.0 // indirect
|
||||
github.com/gogo/protobuf v1.3.2 // indirect
|
||||
github.com/google/gnostic-models v0.7.0 // indirect
|
||||
github.com/google/go-cmp v0.7.0 // indirect
|
||||
github.com/google/uuid v1.6.0 // indirect
|
||||
github.com/josharian/intern v1.0.0 // indirect
|
||||
github.com/json-iterator/go v1.1.12 // indirect
|
||||
github.com/mailru/easyjson v0.7.7 // indirect
|
||||
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
|
||||
github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect
|
||||
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
|
||||
github.com/pmezard/go-difflib v1.0.0 // indirect
|
||||
github.com/prometheus/client_golang v1.23.2 // indirect
|
||||
github.com/prometheus/client_model v0.6.2 // indirect
|
||||
github.com/prometheus/common v0.66.1 // indirect
|
||||
github.com/prometheus/procfs v0.16.1 // indirect
|
||||
github.com/spf13/pflag v1.0.9 // indirect
|
||||
github.com/x448/float16 v0.8.4 // indirect
|
||||
go.yaml.in/yaml/v2 v2.4.3 // indirect
|
||||
go.yaml.in/yaml/v3 v3.0.4 // indirect
|
||||
golang.org/x/net v0.47.0 // indirect
|
||||
golang.org/x/oauth2 v0.30.0 // indirect
|
||||
golang.org/x/sys v0.38.0 // indirect
|
||||
golang.org/x/term v0.37.0 // indirect
|
||||
golang.org/x/text v0.31.0 // indirect
|
||||
golang.org/x/time v0.9.0 // indirect
|
||||
google.golang.org/protobuf v1.36.8 // indirect
|
||||
gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect
|
||||
gopkg.in/inf.v0 v0.9.1 // indirect
|
||||
gopkg.in/yaml.v3 v3.0.1 // indirect
|
||||
k8s.io/apiextensions-apiserver v0.34.3 // indirect
|
||||
k8s.io/klog/v2 v2.130.1 // indirect
|
||||
k8s.io/kube-openapi v0.0.0-20250910181357-589584f1c912 // indirect
|
||||
sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 // indirect
|
||||
sigs.k8s.io/randfill v1.0.0 // indirect
|
||||
sigs.k8s.io/structured-merge-diff/v6 v6.3.2-0.20260122202528-d9cc6641c482 // indirect
|
||||
sigs.k8s.io/yaml v1.6.0 // indirect
|
||||
)
|
||||
|
|
@ -1,571 +0,0 @@
|
|||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
"k8s.io/client-go/rest"
|
||||
"k8s.io/client-go/tools/leaderelection"
|
||||
"k8s.io/client-go/tools/leaderelection/resourcelock"
|
||||
|
||||
"sigs.k8s.io/controller-runtime/pkg/envtest"
|
||||
)
|
||||
|
||||
var (
|
||||
testEnv *envtest.Environment
|
||||
testConfig *rest.Config
|
||||
)
|
||||
|
||||
type electionStateNoRelease struct {
|
||||
cancel context.CancelFunc
|
||||
elected chan struct{}
|
||||
lost chan struct{}
|
||||
}
|
||||
|
||||
func TestMain(m *testing.M) {
|
||||
testEnv = &envtest.Environment{}
|
||||
|
||||
var err error
|
||||
testConfig, err = testEnv.Start()
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "failed to start envtest: %v\n", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
// Set up global client for the wrapper
|
||||
client, err := kubernetes.NewForConfig(testConfig)
|
||||
if err != nil {
|
||||
fmt.Fprintf(os.Stderr, "failed to create clientset: %v\n", err)
|
||||
testEnv.Stop()
|
||||
os.Exit(1)
|
||||
}
|
||||
clientMutex.Lock()
|
||||
globalClient = client
|
||||
clientMutex.Unlock()
|
||||
|
||||
code := m.Run()
|
||||
|
||||
testEnv.Stop()
|
||||
os.Exit(code)
|
||||
}
|
||||
|
||||
func runElectionWithoutRelease(namespace, leaseName, identity string,
|
||||
leaseDurationSec, renewDeadlineSec, retryPeriodSec int) (*electionStateNoRelease, error) {
|
||||
if err := ensureClientInitialized(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
state := &electionStateNoRelease{
|
||||
cancel: cancel,
|
||||
elected: make(chan struct{}),
|
||||
lost: make(chan struct{}),
|
||||
}
|
||||
|
||||
lock := &resourcelock.LeaseLock{
|
||||
LeaseMeta: metav1.ObjectMeta{
|
||||
Name: leaseName,
|
||||
Namespace: namespace,
|
||||
},
|
||||
Client: globalClient.CoordinationV1(),
|
||||
LockConfig: resourcelock.ResourceLockConfig{
|
||||
Identity: identity,
|
||||
},
|
||||
}
|
||||
|
||||
le, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
|
||||
Lock: lock,
|
||||
LeaseDuration: time.Duration(leaseDurationSec) * time.Second,
|
||||
RenewDeadline: time.Duration(renewDeadlineSec) * time.Second,
|
||||
RetryPeriod: time.Duration(retryPeriodSec) * time.Second,
|
||||
ReleaseOnCancel: false,
|
||||
Callbacks: leaderelection.LeaderCallbacks{
|
||||
OnStartedLeading: func(ctx context.Context) {
|
||||
close(state.elected)
|
||||
<-ctx.Done()
|
||||
},
|
||||
OnStoppedLeading: func() {
|
||||
close(state.lost)
|
||||
},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
cancel()
|
||||
return nil, fmt.Errorf("failed to create leader elector: %w", err)
|
||||
}
|
||||
|
||||
go le.Run(ctx)
|
||||
return state, nil
|
||||
}
|
||||
|
||||
// TestSingleLeaderElection verifies a single candidate becomes leader.
|
||||
func TestSingleLeaderElection(t *testing.T) {
|
||||
ns := "default"
|
||||
lease := "single-election-test"
|
||||
identity := "node-1:8080"
|
||||
|
||||
err := runElection(ns, lease, identity, 5, 4, 1)
|
||||
if err != nil {
|
||||
t.Fatalf("runElection failed: %v", err)
|
||||
}
|
||||
|
||||
// Wait for elected
|
||||
key := electionKey(ns, lease)
|
||||
electionMutex.Lock()
|
||||
state := elections[key]
|
||||
electionMutex.Unlock()
|
||||
|
||||
select {
|
||||
case <-state.elected:
|
||||
// success
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("timed out waiting for election")
|
||||
}
|
||||
|
||||
// Verify holder via getHolder
|
||||
holder, transitions, err := getHolder(ns, lease)
|
||||
if err != nil {
|
||||
t.Fatalf("getHolder failed: %v", err)
|
||||
}
|
||||
if holder != identity {
|
||||
t.Errorf("expected holder %q, got %q", identity, holder)
|
||||
}
|
||||
// First election — transitions should be 0 or 1
|
||||
if transitions < 0 {
|
||||
t.Errorf("expected non-negative transitions, got %d", transitions)
|
||||
}
|
||||
|
||||
// Cancel the election
|
||||
electionMutex.Lock()
|
||||
state = elections[key]
|
||||
electionMutex.Unlock()
|
||||
state.cancel()
|
||||
|
||||
select {
|
||||
case <-state.lost:
|
||||
// success
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatal("timed out waiting for election loss after cancel")
|
||||
}
|
||||
}
|
||||
|
||||
// TestLeaderEpoch verifies leaseTransitions increments across elections.
|
||||
func TestLeaderEpoch(t *testing.T) {
|
||||
ns := "default"
|
||||
lease := "epoch-test"
|
||||
|
||||
// First election
|
||||
err := runElection(ns, lease, "node-epoch-1:8080", 5, 4, 1)
|
||||
if err != nil {
|
||||
t.Fatalf("first runElection failed: %v", err)
|
||||
}
|
||||
|
||||
key := electionKey(ns, lease)
|
||||
electionMutex.Lock()
|
||||
state1 := elections[key]
|
||||
electionMutex.Unlock()
|
||||
|
||||
select {
|
||||
case <-state1.elected:
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("timed out on first election")
|
||||
}
|
||||
|
||||
_, trans1, _ := getHolder(ns, lease)
|
||||
|
||||
// Cancel first election and wait for loss
|
||||
state1.cancel()
|
||||
select {
|
||||
case <-state1.lost:
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatal("timed out waiting for first election loss")
|
||||
}
|
||||
|
||||
// Wait for lease to expire / be released
|
||||
time.Sleep(2 * time.Second)
|
||||
|
||||
// Second election
|
||||
err = runElection(ns, lease, "node-epoch-2:8080", 5, 4, 1)
|
||||
if err != nil {
|
||||
t.Fatalf("second runElection failed: %v", err)
|
||||
}
|
||||
|
||||
electionMutex.Lock()
|
||||
state2 := elections[key]
|
||||
electionMutex.Unlock()
|
||||
|
||||
select {
|
||||
case <-state2.elected:
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("timed out on second election")
|
||||
}
|
||||
|
||||
_, trans2, _ := getHolder(ns, lease)
|
||||
if trans2 <= trans1 {
|
||||
t.Errorf("expected transitions to increment: first=%d, second=%d", trans1, trans2)
|
||||
}
|
||||
|
||||
state2.cancel()
|
||||
select {
|
||||
case <-state2.lost:
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatal("timed out waiting for second election loss")
|
||||
}
|
||||
}
|
||||
|
||||
// TestSequentialLeadershipHandoff tests that a second candidate can acquire
|
||||
// leadership after the first one releases it.
|
||||
func TestSequentialLeadershipHandoff(t *testing.T) {
|
||||
ns := "default"
|
||||
lease := "two-candidate-test"
|
||||
|
||||
err1 := runElection(ns, lease, "candidate-a:8080", 5, 4, 1)
|
||||
if err1 != nil {
|
||||
t.Fatalf("first runElection failed: %v", err1)
|
||||
}
|
||||
|
||||
key := electionKey(ns, lease)
|
||||
electionMutex.Lock()
|
||||
stateA := elections[key]
|
||||
electionMutex.Unlock()
|
||||
|
||||
// Wait for first candidate to win
|
||||
select {
|
||||
case <-stateA.elected:
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("timed out waiting for first candidate")
|
||||
}
|
||||
|
||||
// Verify holder is candidate-a
|
||||
holder, _, err := getHolder(ns, lease)
|
||||
if err != nil {
|
||||
t.Fatalf("getHolder failed: %v", err)
|
||||
}
|
||||
if holder != "candidate-a:8080" {
|
||||
t.Errorf("expected candidate-a, got %q", holder)
|
||||
}
|
||||
|
||||
// Cancel candidate-a
|
||||
stateA.cancel()
|
||||
select {
|
||||
case <-stateA.lost:
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatal("timed out waiting for candidate-a loss")
|
||||
}
|
||||
|
||||
// Wait for lease to expire
|
||||
time.Sleep(2 * time.Second)
|
||||
|
||||
// Start candidate-b
|
||||
err2 := runElection(ns, lease, "candidate-b:8080", 5, 4, 1)
|
||||
if err2 != nil {
|
||||
t.Fatalf("second runElection failed: %v", err2)
|
||||
}
|
||||
|
||||
electionMutex.Lock()
|
||||
stateB := elections[key]
|
||||
electionMutex.Unlock()
|
||||
|
||||
select {
|
||||
case <-stateB.elected:
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("timed out waiting for candidate-b")
|
||||
}
|
||||
|
||||
holder, _, err = getHolder(ns, lease)
|
||||
if err != nil {
|
||||
t.Fatalf("getHolder after takeover failed: %v", err)
|
||||
}
|
||||
if holder != "candidate-b:8080" {
|
||||
t.Errorf("expected candidate-b, got %q", holder)
|
||||
}
|
||||
|
||||
stateB.cancel()
|
||||
select {
|
||||
case <-stateB.lost:
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatal("timed out waiting for candidate-b loss")
|
||||
}
|
||||
}
|
||||
|
||||
// TestConcurrentCandidateElection starts two candidates simultaneously and
|
||||
// verifies that exactly one wins leadership.
|
||||
func TestConcurrentCandidateElection(t *testing.T) {
|
||||
ns := "default"
|
||||
lease := "concurrent-election-test"
|
||||
|
||||
type result struct {
|
||||
identity string
|
||||
elected bool
|
||||
}
|
||||
|
||||
candidates := []string{"candidate-a:8080", "candidate-b:8080"}
|
||||
results := make(chan result, len(candidates))
|
||||
|
||||
lock := func(identity string) *resourcelock.LeaseLock {
|
||||
return &resourcelock.LeaseLock{
|
||||
LeaseMeta: metav1.ObjectMeta{
|
||||
Name: lease,
|
||||
Namespace: ns,
|
||||
},
|
||||
Client: globalClient.CoordinationV1(),
|
||||
LockConfig: resourcelock.ResourceLockConfig{
|
||||
Identity: identity,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for _, id := range candidates {
|
||||
wg.Add(1)
|
||||
go func(identity string) {
|
||||
defer wg.Done()
|
||||
|
||||
// Short timeout: enough for one to acquire, but the loser
|
||||
// times out before the winner's lease could expire.
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 8*time.Second)
|
||||
defer cancel()
|
||||
|
||||
elected := make(chan struct{})
|
||||
le, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
|
||||
Lock: lock(identity),
|
||||
LeaseDuration: 5 * time.Second,
|
||||
RenewDeadline: 3 * time.Second,
|
||||
RetryPeriod: 1 * time.Second,
|
||||
ReleaseOnCancel: true,
|
||||
Callbacks: leaderelection.LeaderCallbacks{
|
||||
OnStartedLeading: func(ctx context.Context) {
|
||||
close(elected)
|
||||
<-ctx.Done()
|
||||
},
|
||||
OnStoppedLeading: func() {},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
t.Errorf("NewLeaderElector(%s): %v", identity, err)
|
||||
return
|
||||
}
|
||||
|
||||
go le.Run(ctx)
|
||||
|
||||
select {
|
||||
case <-elected:
|
||||
results <- result{identity, true}
|
||||
// Keep holding until context expires (8s total).
|
||||
// Winner does NOT release early, so loser cannot
|
||||
// re-acquire within its own 8s window.
|
||||
<-ctx.Done()
|
||||
case <-ctx.Done():
|
||||
results <- result{identity, false}
|
||||
}
|
||||
}(id)
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
close(results)
|
||||
|
||||
winners := 0
|
||||
for r := range results {
|
||||
if r.elected {
|
||||
winners++
|
||||
t.Logf("winner: %s", r.identity)
|
||||
}
|
||||
}
|
||||
|
||||
if winners != 1 {
|
||||
t.Fatalf("expected exactly 1 winner, got %d", winners)
|
||||
}
|
||||
}
|
||||
|
||||
// TestCancelElection tests that cancelling an election makes WaitLost return.
|
||||
func TestCancelElection(t *testing.T) {
|
||||
ns := "default"
|
||||
lease := "cancel-test"
|
||||
|
||||
err := runElection(ns, lease, "cancel-node:8080", 5, 4, 1)
|
||||
if err != nil {
|
||||
t.Fatalf("runElection failed: %v", err)
|
||||
}
|
||||
|
||||
key := electionKey(ns, lease)
|
||||
electionMutex.Lock()
|
||||
state := elections[key]
|
||||
electionMutex.Unlock()
|
||||
|
||||
// Wait for elected
|
||||
select {
|
||||
case <-state.elected:
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("timed out waiting for election")
|
||||
}
|
||||
|
||||
// Cancel
|
||||
state.cancel()
|
||||
|
||||
// WaitLost should return promptly
|
||||
select {
|
||||
case <-state.lost:
|
||||
// success
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatal("WaitLost did not return after cancel")
|
||||
}
|
||||
}
|
||||
|
||||
// TestGetHolderDuringElection verifies getHolder works while election is active.
|
||||
func TestGetHolderDuringElection(t *testing.T) {
|
||||
ns := "default"
|
||||
lease := "active-get-holder-test"
|
||||
identity := "active-node:8080"
|
||||
|
||||
err := runElection(ns, lease, identity, 5, 4, 1)
|
||||
if err != nil {
|
||||
t.Fatalf("runElection failed: %v", err)
|
||||
}
|
||||
|
||||
key := electionKey(ns, lease)
|
||||
electionMutex.Lock()
|
||||
state := elections[key]
|
||||
electionMutex.Unlock()
|
||||
|
||||
select {
|
||||
case <-state.elected:
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("timed out waiting for election")
|
||||
}
|
||||
|
||||
// Concurrent getHolder calls during active election
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 5; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
holder, _, err := getHolder(ns, lease)
|
||||
if err != nil {
|
||||
t.Errorf("getHolder during election failed: %v", err)
|
||||
return
|
||||
}
|
||||
if holder != identity {
|
||||
t.Errorf("expected %q, got %q", identity, holder)
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
|
||||
state.cancel()
|
||||
<-state.lost
|
||||
}
|
||||
|
||||
// TestGetHolderReturnsEmptyAfterLeaderDeath verifies that after a leader stops
|
||||
// renewing its lease without releasing it, getHolder returns an empty holder
|
||||
// once the lease expires. This is the integration-level counterpart to the
|
||||
// unit test TestGetHolderReturnsEmptyForExpiredLease.
|
||||
func TestGetHolderReturnsEmptyAfterLeaderDeath(t *testing.T) {
|
||||
ns := "default"
|
||||
lease := "expired-leader-test"
|
||||
identity := "doomed-leader:8080"
|
||||
|
||||
// Acquire leadership without ReleaseOnCancel so canceling simulates a dead
|
||||
// leader that stops renewing and leaves the old holder until expiry.
|
||||
state, err := runElectionWithoutRelease(ns, lease, identity, 5, 4, 1)
|
||||
if err != nil {
|
||||
t.Fatalf("runElection failed: %v", err)
|
||||
}
|
||||
|
||||
select {
|
||||
case <-state.elected:
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("timed out waiting for election")
|
||||
}
|
||||
|
||||
// Verify holder while active.
|
||||
holder, _, err := getHolder(ns, lease)
|
||||
if err != nil {
|
||||
t.Fatalf("getHolder (active) failed: %v", err)
|
||||
}
|
||||
if holder != identity {
|
||||
t.Fatalf("expected active holder %q, got %q", identity, holder)
|
||||
}
|
||||
|
||||
// Simulate leader death: stop renewing without explicitly releasing.
|
||||
state.cancel()
|
||||
select {
|
||||
case <-state.lost:
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatal("timed out waiting for loss")
|
||||
}
|
||||
|
||||
// Wait for the lease to expire (leaseDuration=5s, add margin).
|
||||
time.Sleep(7 * time.Second)
|
||||
|
||||
// After expiry, getHolder must return empty holder so that the
|
||||
// supervisor will attempt acquisition.
|
||||
holder, _, err = getHolder(ns, lease)
|
||||
if err != nil {
|
||||
t.Fatalf("getHolder (expired) failed: %v", err)
|
||||
}
|
||||
if holder != "" {
|
||||
t.Errorf("expected empty holder after lease expiry, got %q", holder)
|
||||
}
|
||||
}
|
||||
|
||||
// TestFailoverAfterLeaderDeath verifies that a new candidate can acquire
|
||||
// leadership after the previous leader dies and its lease expires.
|
||||
func TestFailoverAfterLeaderDeath(t *testing.T) {
|
||||
ns := "default"
|
||||
lease := "failover-test"
|
||||
|
||||
// First leader acquires without ReleaseOnCancel so canceling leaves the
|
||||
// old holder in place until the lease naturally expires.
|
||||
state1, err := runElectionWithoutRelease(ns, lease, "leader-1:8080", 5, 4, 1)
|
||||
if err != nil {
|
||||
t.Fatalf("first runElection failed: %v", err)
|
||||
}
|
||||
|
||||
select {
|
||||
case <-state1.elected:
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("timed out waiting for first election")
|
||||
}
|
||||
|
||||
// Simulate crash: cancel without release, wait for expiry.
|
||||
state1.cancel()
|
||||
<-state1.lost
|
||||
time.Sleep(7 * time.Second)
|
||||
|
||||
// Second candidate should be able to acquire.
|
||||
err = runElection(ns, lease, "leader-2:8080", 5, 4, 1)
|
||||
if err != nil {
|
||||
t.Fatalf("second runElection failed: %v", err)
|
||||
}
|
||||
|
||||
key := electionKey(ns, lease)
|
||||
electionMutex.Lock()
|
||||
state2 := elections[key]
|
||||
electionMutex.Unlock()
|
||||
|
||||
select {
|
||||
case <-state2.elected:
|
||||
// success — failover worked
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Fatal("second candidate failed to acquire after leader death")
|
||||
}
|
||||
|
||||
holder, _, err := getHolder(ns, lease)
|
||||
if err != nil {
|
||||
t.Fatalf("getHolder after failover failed: %v", err)
|
||||
}
|
||||
if holder != "leader-2:8080" {
|
||||
t.Errorf("expected new leader %q, got %q", "leader-2:8080", holder)
|
||||
}
|
||||
|
||||
state2.cancel()
|
||||
<-state2.lost
|
||||
}
|
||||
|
|
@ -1,489 +0,0 @@
|
|||
package main
|
||||
|
||||
/*
|
||||
#include <stdint.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
|
||||
// Trampoline to invoke C/C++ callback safely from Go via cgo.
|
||||
typedef void (*holder_change_cb_t)(void* ctx,
|
||||
const char* holder, size_t holderSize,
|
||||
int64_t leaseTransitions);
|
||||
|
||||
static inline void call_holder_change_cb(holder_change_cb_t func, void* ctx,
|
||||
const char* holder, size_t holderSize,
|
||||
int64_t leaseTransitions) {
|
||||
func(ctx, holder, holderSize, leaseTransitions);
|
||||
}
|
||||
*/
|
||||
import "C"
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
"unsafe"
|
||||
|
||||
coordinationv1 "k8s.io/api/coordination/v1"
|
||||
apierrors "k8s.io/apimachinery/pkg/api/errors"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/watch"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
"k8s.io/client-go/rest"
|
||||
"k8s.io/client-go/tools/clientcmd"
|
||||
"k8s.io/client-go/tools/leaderelection"
|
||||
"k8s.io/client-go/tools/leaderelection/resourcelock"
|
||||
)
|
||||
|
||||
// electionState holds the runtime state for a single leader election.
|
||||
type electionState struct {
|
||||
cancel context.CancelFunc
|
||||
elected chan struct{} // closed when OnStartedLeading fires
|
||||
lost chan struct{} // closed when OnStoppedLeading fires
|
||||
err error // set before lost is closed, if any
|
||||
transitions int64 // set before elected is closed
|
||||
}
|
||||
|
||||
// watchState holds the runtime state for a single Lease watch.
|
||||
type watchState struct {
|
||||
cancel context.CancelFunc
|
||||
}
|
||||
|
||||
var (
|
||||
globalClient kubernetes.Interface
|
||||
clientMutex sync.Mutex
|
||||
initClientFn = initClient
|
||||
|
||||
elections = make(map[string]*electionState)
|
||||
electionMutex sync.Mutex
|
||||
|
||||
watches = make(map[string]*watchState)
|
||||
watchMutex sync.Mutex
|
||||
)
|
||||
|
||||
func electionKey(namespace, leaseName string) string {
|
||||
return namespace + "/" + leaseName
|
||||
}
|
||||
|
||||
func ensureClientInitialized() error {
|
||||
clientMutex.Lock()
|
||||
initialized := globalClient != nil
|
||||
clientMutex.Unlock()
|
||||
if initialized {
|
||||
return nil
|
||||
}
|
||||
return initClientFn()
|
||||
}
|
||||
|
||||
// initClient creates the K8s clientset from in-cluster config or KUBECONFIG.
|
||||
func initClient() error {
|
||||
clientMutex.Lock()
|
||||
defer clientMutex.Unlock()
|
||||
if globalClient != nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
config, err := rest.InClusterConfig()
|
||||
if err != nil {
|
||||
// Fall back to KUBECONFIG
|
||||
kubeconfig := os.Getenv("KUBECONFIG")
|
||||
if kubeconfig == "" {
|
||||
home := os.Getenv("HOME")
|
||||
if home != "" {
|
||||
kubeconfig = home + "/.kube/config"
|
||||
}
|
||||
}
|
||||
config, err = clientcmd.BuildConfigFromFlags("", kubeconfig)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to build k8s config: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
client, err := kubernetes.NewForConfig(config)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create k8s clientset: %w", err)
|
||||
}
|
||||
globalClient = client
|
||||
return nil
|
||||
}
|
||||
|
||||
// runElection starts a leader election goroutine for the given namespace/leaseName.
|
||||
func runElection(namespace, leaseName, identity string,
|
||||
leaseDurationSec, renewDeadlineSec, retryPeriodSec int) error {
|
||||
if err := ensureClientInitialized(); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
key := electionKey(namespace, leaseName)
|
||||
|
||||
electionMutex.Lock()
|
||||
if _, exists := elections[key]; exists {
|
||||
electionMutex.Unlock()
|
||||
return fmt.Errorf("election already running for %s", key)
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
state := &electionState{
|
||||
cancel: cancel,
|
||||
elected: make(chan struct{}),
|
||||
lost: make(chan struct{}),
|
||||
}
|
||||
elections[key] = state
|
||||
electionMutex.Unlock()
|
||||
|
||||
lock := &resourcelock.LeaseLock{
|
||||
LeaseMeta: metav1.ObjectMeta{
|
||||
Name: leaseName,
|
||||
Namespace: namespace,
|
||||
},
|
||||
Client: globalClient.CoordinationV1(),
|
||||
LockConfig: resourcelock.ResourceLockConfig{
|
||||
Identity: identity,
|
||||
},
|
||||
}
|
||||
|
||||
le, err := leaderelection.NewLeaderElector(leaderelection.LeaderElectionConfig{
|
||||
Lock: lock,
|
||||
LeaseDuration: time.Duration(leaseDurationSec) * time.Second,
|
||||
RenewDeadline: time.Duration(renewDeadlineSec) * time.Second,
|
||||
RetryPeriod: time.Duration(retryPeriodSec) * time.Second,
|
||||
ReleaseOnCancel: true,
|
||||
Callbacks: leaderelection.LeaderCallbacks{
|
||||
OnStartedLeading: func(ctx context.Context) {
|
||||
_, transitions, err := getHolder(namespace, leaseName)
|
||||
if err == nil {
|
||||
state.transitions = transitions
|
||||
}
|
||||
close(state.elected)
|
||||
// Block until context is cancelled (leadership lost or explicit cancel)
|
||||
<-ctx.Done()
|
||||
},
|
||||
OnStoppedLeading: func() {
|
||||
close(state.lost)
|
||||
// Auto-cleanup: remove from map so the same key can be reused.
|
||||
electionMutex.Lock()
|
||||
if elections[key] == state {
|
||||
delete(elections, key)
|
||||
}
|
||||
electionMutex.Unlock()
|
||||
},
|
||||
},
|
||||
})
|
||||
if err != nil {
|
||||
electionMutex.Lock()
|
||||
delete(elections, key)
|
||||
electionMutex.Unlock()
|
||||
cancel()
|
||||
return fmt.Errorf("failed to create leader elector: %w", err)
|
||||
}
|
||||
|
||||
go le.Run(ctx)
|
||||
return nil
|
||||
}
|
||||
|
||||
// getHolder reads the current Lease holder identity and transitions.
|
||||
func getHolder(namespace, leaseName string) (string, int64, error) {
|
||||
if err := ensureClientInitialized(); err != nil {
|
||||
return "", 0, err
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
lease, err := globalClient.CoordinationV1().Leases(namespace).Get(ctx, leaseName, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
return "", 0, fmt.Errorf("failed to get lease: %w", err)
|
||||
}
|
||||
|
||||
holder := ""
|
||||
if lease.Spec.HolderIdentity != nil {
|
||||
holder = *lease.Spec.HolderIdentity
|
||||
}
|
||||
transitions := int64(0)
|
||||
if lease.Spec.LeaseTransitions != nil {
|
||||
transitions = int64(*lease.Spec.LeaseTransitions)
|
||||
}
|
||||
|
||||
// Treat expired leases as having no holder so that the C++ supervisor
|
||||
// will attempt acquisition instead of going to standby.
|
||||
if holder != "" && lease.Spec.RenewTime != nil && lease.Spec.LeaseDurationSeconds != nil {
|
||||
expiry := lease.Spec.RenewTime.Time.Add(time.Duration(*lease.Spec.LeaseDurationSeconds) * time.Second)
|
||||
if time.Now().After(expiry) {
|
||||
holder = ""
|
||||
}
|
||||
}
|
||||
return holder, transitions, nil
|
||||
}
|
||||
|
||||
//export K8sLeaseInit
|
||||
func K8sLeaseInit(errMsg **C.char) C.int {
|
||||
if err := ensureClientInitialized(); err != nil {
|
||||
*errMsg = C.CString(err.Error())
|
||||
return -1
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
//export K8sLeaseRunElection
|
||||
func K8sLeaseRunElection(
|
||||
ns, leaseName, identity *C.char,
|
||||
leaseDurationSec, renewDeadlineSec, retryPeriodSec C.int,
|
||||
errMsg **C.char,
|
||||
) C.int {
|
||||
nsStr := C.GoString(ns)
|
||||
ln := C.GoString(leaseName)
|
||||
id := C.GoString(identity)
|
||||
|
||||
err := runElection(nsStr, ln, id,
|
||||
int(leaseDurationSec), int(renewDeadlineSec), int(retryPeriodSec))
|
||||
if err != nil {
|
||||
*errMsg = C.CString(err.Error())
|
||||
return -1
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
//export K8sLeaseWaitElected
|
||||
func K8sLeaseWaitElected(
|
||||
ns, leaseName *C.char,
|
||||
timeoutSec C.int,
|
||||
leaseTransitions *C.longlong,
|
||||
errMsg **C.char,
|
||||
) C.int {
|
||||
key := electionKey(C.GoString(ns), C.GoString(leaseName))
|
||||
|
||||
electionMutex.Lock()
|
||||
state, exists := elections[key]
|
||||
electionMutex.Unlock()
|
||||
|
||||
if !exists {
|
||||
*errMsg = C.CString("no election running for " + key)
|
||||
return -1
|
||||
}
|
||||
|
||||
timeout := time.Duration(timeoutSec) * time.Second
|
||||
|
||||
// Wait for elected, lost, or timeout
|
||||
select {
|
||||
case <-state.elected:
|
||||
*leaseTransitions = C.longlong(state.transitions)
|
||||
return 0
|
||||
case <-state.lost:
|
||||
*errMsg = C.CString("election lost before becoming leader")
|
||||
return -1
|
||||
case <-time.After(timeout):
|
||||
state.cancel()
|
||||
<-state.lost
|
||||
*errMsg = C.CString("election timed out after " + fmt.Sprintf("%d", int(timeoutSec)) + "s")
|
||||
return -1
|
||||
}
|
||||
}
|
||||
|
||||
//export K8sLeaseWaitLost
|
||||
func K8sLeaseWaitLost(
|
||||
ns, leaseName *C.char,
|
||||
errMsg **C.char,
|
||||
) C.int {
|
||||
key := electionKey(C.GoString(ns), C.GoString(leaseName))
|
||||
|
||||
electionMutex.Lock()
|
||||
state, exists := elections[key]
|
||||
electionMutex.Unlock()
|
||||
|
||||
if !exists {
|
||||
// Already cleaned up by OnStoppedLeading — election is over.
|
||||
return 0
|
||||
}
|
||||
|
||||
<-state.lost
|
||||
|
||||
if state.err != nil {
|
||||
*errMsg = C.CString(state.err.Error())
|
||||
return -1
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
//export K8sLeaseCancelElection
|
||||
func K8sLeaseCancelElection(
|
||||
ns, leaseName *C.char,
|
||||
errMsg **C.char,
|
||||
) C.int {
|
||||
key := electionKey(C.GoString(ns), C.GoString(leaseName))
|
||||
|
||||
electionMutex.Lock()
|
||||
state, exists := elections[key]
|
||||
electionMutex.Unlock()
|
||||
|
||||
if !exists {
|
||||
// Idempotent — no error if no election
|
||||
return 0
|
||||
}
|
||||
|
||||
state.cancel()
|
||||
return 0
|
||||
}
|
||||
|
||||
//export K8sLeaseGetHolder
|
||||
func K8sLeaseGetHolder(
|
||||
ns, leaseName *C.char,
|
||||
holderIdentity **C.char,
|
||||
leaseTransitions *C.longlong,
|
||||
errMsg **C.char,
|
||||
) C.int {
|
||||
nsStr := C.GoString(ns)
|
||||
ln := C.GoString(leaseName)
|
||||
|
||||
holder, transitions, err := getHolder(nsStr, ln)
|
||||
if err != nil {
|
||||
if apierrors.IsNotFound(err) {
|
||||
*holderIdentity = nil
|
||||
*leaseTransitions = 0
|
||||
return 1
|
||||
}
|
||||
errStr := err.Error()
|
||||
*errMsg = C.CString(errStr)
|
||||
return -1
|
||||
}
|
||||
|
||||
if holder == "" {
|
||||
*holderIdentity = nil
|
||||
} else {
|
||||
*holderIdentity = C.CString(holder)
|
||||
}
|
||||
*leaseTransitions = C.longlong(transitions)
|
||||
return 0
|
||||
}
|
||||
|
||||
//export K8sLeaseWatchHolder
|
||||
func K8sLeaseWatchHolder(
|
||||
ns, leaseName *C.char,
|
||||
callbackCtx unsafe.Pointer,
|
||||
callbackFunc C.holder_change_cb_t,
|
||||
errMsg **C.char,
|
||||
) C.int {
|
||||
nsStr := C.GoString(ns)
|
||||
ln := C.GoString(leaseName)
|
||||
key := electionKey(nsStr, ln)
|
||||
|
||||
if callbackFunc == nil {
|
||||
*errMsg = C.CString("callback function is nil")
|
||||
return -1
|
||||
}
|
||||
if err := ensureClientInitialized(); err != nil {
|
||||
*errMsg = C.CString(err.Error())
|
||||
return -1
|
||||
}
|
||||
|
||||
watchMutex.Lock()
|
||||
if _, exists := watches[key]; exists {
|
||||
watchMutex.Unlock()
|
||||
*errMsg = C.CString("watch already running for " + key)
|
||||
return -1
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
watches[key] = &watchState{cancel: cancel}
|
||||
watchMutex.Unlock()
|
||||
|
||||
go func() {
|
||||
defer func() {
|
||||
watchMutex.Lock()
|
||||
delete(watches, key)
|
||||
watchMutex.Unlock()
|
||||
}()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
watcher, err := globalClient.CoordinationV1().Leases(nsStr).Watch(ctx, metav1.ListOptions{
|
||||
FieldSelector: "metadata.name=" + ln,
|
||||
})
|
||||
if err != nil {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
time.Sleep(time.Second)
|
||||
continue
|
||||
}
|
||||
}
|
||||
|
||||
for event := range watcher.ResultChan() {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
watcher.Stop()
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
if event.Type == watch.Modified || event.Type == watch.Added {
|
||||
lease, ok := event.Object.(*coordinationv1.Lease)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
holder := ""
|
||||
if lease.Spec.HolderIdentity != nil {
|
||||
holder = *lease.Spec.HolderIdentity
|
||||
}
|
||||
transitions := int64(0)
|
||||
if lease.Spec.LeaseTransitions != nil {
|
||||
transitions = int64(*lease.Spec.LeaseTransitions)
|
||||
}
|
||||
|
||||
var holderPtr *C.char
|
||||
var holderSize C.size_t
|
||||
if holder != "" {
|
||||
holderPtr = C.CString(holder)
|
||||
holderSize = C.size_t(len(holder))
|
||||
}
|
||||
|
||||
C.call_holder_change_cb(callbackFunc, callbackCtx,
|
||||
holderPtr, holderSize, C.int64_t(transitions))
|
||||
|
||||
if holderPtr != nil {
|
||||
C.free(unsafe.Pointer(holderPtr))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Watch channel closed — retry unless cancelled
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
time.Sleep(time.Second)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
return 0
|
||||
}
|
||||
|
||||
//export K8sLeaseCancelWatch
|
||||
func K8sLeaseCancelWatch(
|
||||
ns, leaseName *C.char,
|
||||
errMsg **C.char,
|
||||
) C.int {
|
||||
key := electionKey(C.GoString(ns), C.GoString(leaseName))
|
||||
|
||||
watchMutex.Lock()
|
||||
state, exists := watches[key]
|
||||
watchMutex.Unlock()
|
||||
|
||||
if !exists {
|
||||
// Idempotent
|
||||
return 0
|
||||
}
|
||||
|
||||
state.cancel()
|
||||
return 0
|
||||
}
|
||||
|
||||
func main() {}
|
||||
|
|
@ -1,293 +0,0 @@
|
|||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
coordinationv1 "k8s.io/api/coordination/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/client-go/kubernetes"
|
||||
"k8s.io/client-go/kubernetes/fake"
|
||||
"k8s.io/utils/ptr"
|
||||
)
|
||||
|
||||
// swapClient replaces globalClient and returns the old one.
|
||||
func swapClient(newClient kubernetes.Interface) kubernetes.Interface {
|
||||
clientMutex.Lock()
|
||||
defer clientMutex.Unlock()
|
||||
old := globalClient
|
||||
globalClient = newClient
|
||||
return old
|
||||
}
|
||||
|
||||
// TestGetHolderWithFakeClient tests getHolder using a fake K8s clientset.
|
||||
func TestGetHolderWithFakeClient(t *testing.T) {
|
||||
holderID := "node-1:8080"
|
||||
transitions := int32(3)
|
||||
lease := &coordinationv1.Lease{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "test-lease",
|
||||
Namespace: "default",
|
||||
},
|
||||
Spec: coordinationv1.LeaseSpec{
|
||||
HolderIdentity: &holderID,
|
||||
LeaseTransitions: &transitions,
|
||||
},
|
||||
}
|
||||
|
||||
fakeClient := fake.NewSimpleClientset(lease)
|
||||
old := swapClient(fakeClient)
|
||||
defer swapClient(old)
|
||||
|
||||
holder, trans, err := getHolder("default", "test-lease")
|
||||
if err != nil {
|
||||
t.Fatalf("getHolder failed: %v", err)
|
||||
}
|
||||
if holder != holderID {
|
||||
t.Errorf("expected holder %q, got %q", holderID, holder)
|
||||
}
|
||||
if trans != int64(transitions) {
|
||||
t.Errorf("expected transitions %d, got %d", transitions, trans)
|
||||
}
|
||||
}
|
||||
|
||||
// TestGetHolderNotFound tests getHolder when the Lease does not exist.
|
||||
func TestGetHolderNotFound(t *testing.T) {
|
||||
fakeClient := fake.NewSimpleClientset()
|
||||
old := swapClient(fakeClient)
|
||||
defer swapClient(old)
|
||||
|
||||
_, _, err := getHolder("default", "nonexistent")
|
||||
if err == nil {
|
||||
t.Fatal("expected error for nonexistent lease, got nil")
|
||||
}
|
||||
}
|
||||
|
||||
// TestGetHolderEmptyIdentity tests getHolder when holder is nil.
|
||||
func TestGetHolderEmptyIdentity(t *testing.T) {
|
||||
lease := &coordinationv1.Lease{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "empty-lease",
|
||||
Namespace: "default",
|
||||
},
|
||||
Spec: coordinationv1.LeaseSpec{},
|
||||
}
|
||||
fakeClient := fake.NewSimpleClientset(lease)
|
||||
old := swapClient(fakeClient)
|
||||
defer swapClient(old)
|
||||
|
||||
holder, trans, err := getHolder("default", "empty-lease")
|
||||
if err != nil {
|
||||
t.Fatalf("getHolder failed: %v", err)
|
||||
}
|
||||
if holder != "" {
|
||||
t.Errorf("expected empty holder, got %q", holder)
|
||||
}
|
||||
if trans != 0 {
|
||||
t.Errorf("expected 0 transitions, got %d", trans)
|
||||
}
|
||||
}
|
||||
|
||||
// TestGetHolderReturnsEmptyForExpiredLease verifies that getHolder treats a
|
||||
// lease whose renewTime + leaseDuration is in the past as having no holder.
|
||||
// This is critical for failover: when a leader pod dies without releasing the
|
||||
// lease, standbys must see an empty holder so the supervisor attempts
|
||||
// acquisition instead of looping in standby.
|
||||
func TestGetHolderReturnsEmptyForExpiredLease(t *testing.T) {
|
||||
holderID := "dead-leader:8080"
|
||||
leaseDuration := int32(5)
|
||||
transitions := int32(2)
|
||||
expiredRenewTime := metav1.NewMicroTime(time.Now().Add(-10 * time.Second))
|
||||
|
||||
lease := &coordinationv1.Lease{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "expired-lease",
|
||||
Namespace: "default",
|
||||
},
|
||||
Spec: coordinationv1.LeaseSpec{
|
||||
HolderIdentity: &holderID,
|
||||
LeaseDurationSeconds: &leaseDuration,
|
||||
LeaseTransitions: &transitions,
|
||||
RenewTime: &expiredRenewTime,
|
||||
},
|
||||
}
|
||||
|
||||
fakeClient := fake.NewSimpleClientset(lease)
|
||||
old := swapClient(fakeClient)
|
||||
defer swapClient(old)
|
||||
|
||||
holder, trans, err := getHolder("default", "expired-lease")
|
||||
if err != nil {
|
||||
t.Fatalf("getHolder failed: %v", err)
|
||||
}
|
||||
if holder != "" {
|
||||
t.Errorf("expected empty holder for expired lease, got %q", holder)
|
||||
}
|
||||
// Transitions should still be reported even for expired leases.
|
||||
if trans != int64(transitions) {
|
||||
t.Errorf("expected transitions %d, got %d", transitions, trans)
|
||||
}
|
||||
}
|
||||
|
||||
// TestGetHolderReturnsHolderForActiveLease verifies that getHolder returns the
|
||||
// holder identity when the lease is still active (renewTime + leaseDuration is
|
||||
// in the future).
|
||||
func TestGetHolderReturnsHolderForActiveLease(t *testing.T) {
|
||||
holderID := "active-leader:8080"
|
||||
leaseDuration := int32(15)
|
||||
transitions := int32(1)
|
||||
recentRenewTime := metav1.NewMicroTime(time.Now())
|
||||
|
||||
lease := &coordinationv1.Lease{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "active-lease",
|
||||
Namespace: "default",
|
||||
},
|
||||
Spec: coordinationv1.LeaseSpec{
|
||||
HolderIdentity: &holderID,
|
||||
LeaseDurationSeconds: &leaseDuration,
|
||||
LeaseTransitions: &transitions,
|
||||
RenewTime: &recentRenewTime,
|
||||
},
|
||||
}
|
||||
|
||||
fakeClient := fake.NewSimpleClientset(lease)
|
||||
old := swapClient(fakeClient)
|
||||
defer swapClient(old)
|
||||
|
||||
holder, trans, err := getHolder("default", "active-lease")
|
||||
if err != nil {
|
||||
t.Fatalf("getHolder failed: %v", err)
|
||||
}
|
||||
if holder != holderID {
|
||||
t.Errorf("expected holder %q, got %q", holderID, holder)
|
||||
}
|
||||
if trans != int64(transitions) {
|
||||
t.Errorf("expected transitions %d, got %d", transitions, trans)
|
||||
}
|
||||
}
|
||||
|
||||
// TestElectionKeyFormat tests the election key construction.
|
||||
func TestElectionKeyFormat(t *testing.T) {
|
||||
tests := []struct {
|
||||
ns, name, want string
|
||||
}{
|
||||
{"default", "leader", "default/leader"},
|
||||
{"kube-system", "my-lock", "kube-system/my-lock"},
|
||||
{"", "bare", "/bare"},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
got := electionKey(tc.ns, tc.name)
|
||||
if got != tc.want {
|
||||
t.Errorf("electionKey(%q, %q) = %q, want %q", tc.ns, tc.name, got, tc.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestLeaseCRUDWithFakeClient tests basic Lease CRUD via the K8s API.
|
||||
func TestLeaseCRUDWithFakeClient(t *testing.T) {
|
||||
fakeClient := fake.NewSimpleClientset()
|
||||
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
holderID := "node-a:9090"
|
||||
transitions := int32(0)
|
||||
lease := &coordinationv1.Lease{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "crud-test",
|
||||
Namespace: "default",
|
||||
},
|
||||
Spec: coordinationv1.LeaseSpec{
|
||||
HolderIdentity: &holderID,
|
||||
LeaseTransitions: &transitions,
|
||||
},
|
||||
}
|
||||
created, err := fakeClient.CoordinationV1().Leases("default").Create(ctx, lease, metav1.CreateOptions{})
|
||||
if err != nil {
|
||||
t.Fatalf("create lease failed: %v", err)
|
||||
}
|
||||
if *created.Spec.HolderIdentity != holderID {
|
||||
t.Errorf("created holder = %q, want %q", *created.Spec.HolderIdentity, holderID)
|
||||
}
|
||||
|
||||
newHolder := "node-b:9090"
|
||||
newTransitions := int32(1)
|
||||
created.Spec.HolderIdentity = &newHolder
|
||||
created.Spec.LeaseTransitions = &newTransitions
|
||||
updated, err := fakeClient.CoordinationV1().Leases("default").Update(ctx, created, metav1.UpdateOptions{})
|
||||
if err != nil {
|
||||
t.Fatalf("update lease failed: %v", err)
|
||||
}
|
||||
if *updated.Spec.HolderIdentity != newHolder {
|
||||
t.Errorf("updated holder = %q, want %q", *updated.Spec.HolderIdentity, newHolder)
|
||||
}
|
||||
if *updated.Spec.LeaseTransitions != newTransitions {
|
||||
t.Errorf("updated transitions = %d, want %d", *updated.Spec.LeaseTransitions, newTransitions)
|
||||
}
|
||||
|
||||
got, err := fakeClient.CoordinationV1().Leases("default").Get(ctx, "crud-test", metav1.GetOptions{})
|
||||
if err != nil {
|
||||
t.Fatalf("get lease failed: %v", err)
|
||||
}
|
||||
if *got.Spec.HolderIdentity != newHolder {
|
||||
t.Errorf("got holder = %q, want %q", *got.Spec.HolderIdentity, newHolder)
|
||||
}
|
||||
|
||||
err = fakeClient.CoordinationV1().Leases("default").Delete(ctx, "crud-test", metav1.DeleteOptions{})
|
||||
if err != nil {
|
||||
t.Fatalf("delete lease failed: %v", err)
|
||||
}
|
||||
|
||||
_, err = fakeClient.CoordinationV1().Leases("default").Get(ctx, "crud-test", metav1.GetOptions{})
|
||||
if err == nil {
|
||||
t.Fatal("expected error after delete, got nil")
|
||||
}
|
||||
}
|
||||
|
||||
// TestConcurrentGetHolder tests concurrent calls to getHolder.
|
||||
func TestConcurrentGetHolder(t *testing.T) {
|
||||
holderID := "concurrent-node:8080"
|
||||
lease := &coordinationv1.Lease{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: "concurrent-lease",
|
||||
Namespace: "default",
|
||||
},
|
||||
Spec: coordinationv1.LeaseSpec{
|
||||
HolderIdentity: &holderID,
|
||||
LeaseTransitions: ptr.To(int32(5)),
|
||||
},
|
||||
}
|
||||
fakeClient := fake.NewSimpleClientset(lease)
|
||||
old := swapClient(fakeClient)
|
||||
defer swapClient(old)
|
||||
|
||||
const n = 10
|
||||
errCh := make(chan error, n)
|
||||
for i := 0; i < n; i++ {
|
||||
go func() {
|
||||
holder, trans, err := getHolder("default", "concurrent-lease")
|
||||
if err != nil {
|
||||
errCh <- err
|
||||
return
|
||||
}
|
||||
if holder != holderID {
|
||||
errCh <- fmt.Errorf("expected holder %q, got %q", holderID, holder)
|
||||
return
|
||||
}
|
||||
if trans != 5 {
|
||||
errCh <- fmt.Errorf("expected transitions 5, got %d", trans)
|
||||
return
|
||||
}
|
||||
errCh <- nil
|
||||
}()
|
||||
}
|
||||
|
||||
for i := 0; i < n; i++ {
|
||||
if err := <-errCh; err != nil {
|
||||
t.Fatalf("concurrent getHolder failed: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -81,10 +81,6 @@ struct MasterConfig {
|
|||
std::string cxl_path;
|
||||
size_t cxl_size;
|
||||
bool enable_cxl = false;
|
||||
|
||||
// Offload-on-evict: defer LOCAL_DISK offload to eviction time
|
||||
bool offload_on_evict = false;
|
||||
bool offload_force_evict = false;
|
||||
};
|
||||
|
||||
class MasterServiceSupervisorConfig {
|
||||
|
|
@ -144,8 +140,6 @@ class MasterServiceSupervisorConfig {
|
|||
std::string cxl_path = DEFAULT_CXL_PATH;
|
||||
size_t cxl_size = DEFAULT_CXL_SIZE;
|
||||
bool enable_cxl = false;
|
||||
bool offload_on_evict = false;
|
||||
bool offload_force_evict = false;
|
||||
MasterServiceSupervisorConfig() = default;
|
||||
|
||||
// From MasterConfig
|
||||
|
|
@ -161,8 +155,6 @@ class MasterServiceSupervisorConfig {
|
|||
eviction_high_watermark_ratio = config.eviction_high_watermark_ratio;
|
||||
client_live_ttl_sec = config.client_live_ttl_sec;
|
||||
enable_offload = config.enable_offload;
|
||||
offload_on_evict = config.offload_on_evict;
|
||||
offload_force_evict = config.offload_force_evict;
|
||||
rpc_port = static_cast<int>(config.rpc_port);
|
||||
rpc_thread_num = static_cast<size_t>(config.rpc_thread_num);
|
||||
|
||||
|
|
@ -275,8 +267,6 @@ class WrappedMasterServiceConfig {
|
|||
int64_t client_live_ttl_sec = DEFAULT_CLIENT_LIVE_TTL_SEC;
|
||||
bool enable_ha = false;
|
||||
bool enable_offload = false;
|
||||
bool offload_on_evict = false;
|
||||
bool offload_force_evict = false;
|
||||
std::string ha_backend_type = "etcd";
|
||||
std::string ha_backend_connstring;
|
||||
std::string cluster_id = DEFAULT_CLUSTER_ID;
|
||||
|
|
@ -332,8 +322,6 @@ class WrappedMasterServiceConfig {
|
|||
client_live_ttl_sec = config.client_live_ttl_sec;
|
||||
enable_ha = config.enable_ha;
|
||||
enable_offload = config.enable_offload;
|
||||
offload_on_evict = config.offload_on_evict;
|
||||
offload_force_evict = config.offload_force_evict;
|
||||
ha_backend_type = config.ha_backend_type;
|
||||
ha_backend_connstring = config.ha_backend_connstring;
|
||||
if (ha_backend_connstring.empty()) {
|
||||
|
|
@ -412,8 +400,6 @@ class WrappedMasterServiceConfig {
|
|||
enable_ha =
|
||||
true; // This is used in HA mode, so enable_ha should be true
|
||||
enable_offload = config.enable_offload;
|
||||
offload_on_evict = config.offload_on_evict;
|
||||
offload_force_evict = config.offload_force_evict;
|
||||
ha_backend_type = config.ha_backend_type;
|
||||
ha_backend_connstring = config.ha_backend_connstring;
|
||||
if (ha_backend_connstring.empty()) {
|
||||
|
|
@ -750,8 +736,6 @@ class MasterServiceConfig {
|
|||
int64_t client_live_ttl_sec = DEFAULT_CLIENT_LIVE_TTL_SEC;
|
||||
bool enable_ha = false;
|
||||
bool enable_offload = false;
|
||||
bool offload_on_evict = false;
|
||||
bool offload_force_evict = false;
|
||||
std::string ha_backend_type = "etcd";
|
||||
std::string ha_backend_connstring;
|
||||
std::string cluster_id = DEFAULT_CLUSTER_ID;
|
||||
|
|
@ -803,8 +787,6 @@ class MasterServiceConfig {
|
|||
client_live_ttl_sec = config.client_live_ttl_sec;
|
||||
enable_ha = config.enable_ha;
|
||||
enable_offload = config.enable_offload;
|
||||
offload_on_evict = config.offload_on_evict;
|
||||
offload_force_evict = config.offload_force_evict;
|
||||
ha_backend_type = config.ha_backend_type;
|
||||
ha_backend_connstring = config.ha_backend_connstring;
|
||||
cluster_id = config.cluster_id;
|
||||
|
|
|
|||
|
|
@ -1168,14 +1168,6 @@ class MasterService {
|
|||
|
||||
const bool enable_offload_;
|
||||
|
||||
// Offload-on-evict: defer disk offload to eviction time
|
||||
// (config: offload_on_evict)
|
||||
bool offload_on_evict_{false};
|
||||
// Force-evict: allow evicting MEMORY replicas without disk offload when cap
|
||||
// exceeded (config: offload_force_evict, only effective when
|
||||
// offload_on_evict_=true)
|
||||
bool offload_force_evict_{false};
|
||||
|
||||
const std::string ha_backend_type_;
|
||||
|
||||
const std::string ha_backend_connstring_;
|
||||
|
|
|
|||
|
|
@ -109,10 +109,6 @@ DEFINE_validator(eviction_ratio, [](const char* flagname, double value) {
|
|||
DEFINE_bool(enable_ha, false,
|
||||
"Enable high availability, which depends on etcd");
|
||||
DEFINE_bool(enable_offload, false, "Enable offload availability");
|
||||
DEFINE_bool(offload_on_evict, false,
|
||||
"Defer LOCAL_DISK offload to eviction time instead of PutEnd");
|
||||
DEFINE_bool(offload_force_evict, false,
|
||||
"Force-evict objects exceeding offload cap without disk offload");
|
||||
DEFINE_string(ha_backend_type, "etcd",
|
||||
"HA backend type, e.g. etcd | redis | k8s");
|
||||
DEFINE_string(ha_backend_connstring, "",
|
||||
|
|
@ -297,11 +293,6 @@ void InitMasterConf(const mooncake::DefaultConfig& default_config,
|
|||
FLAGS_enable_ha);
|
||||
default_config.GetBool("enable_offload", &master_config.enable_offload,
|
||||
FLAGS_enable_offload);
|
||||
default_config.GetBool("offload_on_evict", &master_config.offload_on_evict,
|
||||
FLAGS_offload_on_evict);
|
||||
default_config.GetBool("offload_force_evict",
|
||||
&master_config.offload_force_evict,
|
||||
FLAGS_offload_force_evict);
|
||||
default_config.GetString("ha_backend_type", &master_config.ha_backend_type,
|
||||
FLAGS_ha_backend_type);
|
||||
default_config.GetString("ha_backend_connstring",
|
||||
|
|
@ -876,8 +867,6 @@ int main(int argc, char* argv[]) {
|
|||
<< master_config.eviction_high_watermark_ratio
|
||||
<< ", enable_ha=" << master_config.enable_ha
|
||||
<< ", enable_offload=" << master_config.enable_offload
|
||||
<< ", offload_on_evict=" << master_config.offload_on_evict
|
||||
<< ", offload_force_evict=" << master_config.offload_force_evict
|
||||
<< ", ha_backend_type=" << master_config.ha_backend_type
|
||||
<< ", ha_backend_connstring=" << ha_backend_connstring
|
||||
<< ", etcd_endpoints=" << master_config.etcd_endpoints
|
||||
|
|
|
|||
|
|
@ -48,13 +48,6 @@ namespace {
|
|||
|
||||
constexpr size_t kUnlimitedSnapshotList = 0;
|
||||
|
||||
// Per-cycle offload cap as a fraction of `offloading_queue_limit_`. Used only
|
||||
// when offload-on-evict mode is active. Defers memory eviction for at most
|
||||
// this fraction of the queue limit per BatchEvict cycle; beyond that, eviction
|
||||
// falls back according to `offload_force_evict_`. A future change may expose
|
||||
// this as a configurable parameter if workloads demand tuning.
|
||||
constexpr double kOffloadCapRatio = 0.5;
|
||||
|
||||
enum class SnapshotCatalogBackendKind {
|
||||
kEmbedded,
|
||||
kRedis,
|
||||
|
|
@ -166,19 +159,6 @@ MasterService::MasterService(const MasterServiceConfig& config)
|
|||
"put_start_discard_timeout_sec");
|
||||
}
|
||||
|
||||
// Offload-on-evict: defer LOCAL_DISK offload to eviction time
|
||||
offload_on_evict_ = enable_offload_ && config.offload_on_evict;
|
||||
if (offload_on_evict_) {
|
||||
LOG(INFO) << "Offload-on-evict mode enabled: DRAM offload to "
|
||||
"LOCAL_DISK will occur at eviction time instead of "
|
||||
"PutEnd";
|
||||
offload_force_evict_ = config.offload_force_evict;
|
||||
if (offload_force_evict_) {
|
||||
LOG(INFO) << "Force-evict enabled: objects exceeding offload "
|
||||
"cap will be evicted without disk offload";
|
||||
}
|
||||
}
|
||||
|
||||
eviction_running_ = true;
|
||||
eviction_thread_ = std::thread(&MasterService::EvictionThreadFunc, this);
|
||||
VLOG(1) << "action=start_eviction_thread";
|
||||
|
|
@ -930,7 +910,7 @@ auto MasterService::PutEnd(const UUID& client_id, const std::string& key,
|
|||
},
|
||||
[](Replica& replica) { replica.mark_complete(); });
|
||||
|
||||
if (enable_offload_ && !offload_on_evict_) {
|
||||
if (enable_offload_) {
|
||||
auto& shard = accessor.GetShard();
|
||||
metadata.VisitReplicas(
|
||||
&Replica::fn_is_completed, [this, &key, &shard](Replica& replica) {
|
||||
|
|
@ -3511,81 +3491,6 @@ void MasterService::BatchEvict(double evict_ratio_target,
|
|||
});
|
||||
};
|
||||
|
||||
// --- Offload-on-evict support ---
|
||||
long offload_queued_this_cycle = 0;
|
||||
long offload_deferred_count = 0;
|
||||
long offload_cap_forced_count = 0; // #keys force-evicted due to cap
|
||||
long offload_push_failed_forced = 0; // #keys force-evicted on push fail
|
||||
const long offload_cap =
|
||||
offload_on_evict_
|
||||
? static_cast<long>(offloading_queue_limit_ * kOffloadCapRatio)
|
||||
: 0;
|
||||
|
||||
auto has_local_disk_replica = [](const ObjectMetadata& metadata) {
|
||||
return metadata.HasReplica(&Replica::fn_is_local_disk_replica);
|
||||
};
|
||||
|
||||
// Returns freed bytes. Returns 0 if offload-queued and no additional
|
||||
// replicas were evicted (all MEMORY replicas of the key are now pinned).
|
||||
auto try_evict_or_offload =
|
||||
[&, this](const std::string& key, ObjectMetadata& metadata,
|
||||
MetadataShardAccessorRW& shard) -> uint64_t {
|
||||
if (!offload_on_evict_) {
|
||||
// Original behavior
|
||||
return metadata.size * evict_replicas(metadata);
|
||||
}
|
||||
|
||||
// LOCAL_DISK replica already exists — safe to delete MEMORY immediately
|
||||
if (has_local_disk_replica(metadata)) {
|
||||
return metadata.size * evict_replicas(metadata);
|
||||
}
|
||||
|
||||
// Force-evict cap: if force_evict enabled and cap reached, force
|
||||
// delete. Warning is aggregated at the end of the cycle to avoid log
|
||||
// flooding.
|
||||
if (offload_force_evict_ && offload_queued_this_cycle >= offload_cap) {
|
||||
offload_cap_forced_count++;
|
||||
return metadata.size * evict_replicas(metadata);
|
||||
}
|
||||
|
||||
// Queue one MEMORY replica for offload; others will be evicted below.
|
||||
bool queued = false;
|
||||
metadata.VisitReplicas(
|
||||
[](const Replica& r) {
|
||||
return r.is_memory_replica() && r.is_completed() &&
|
||||
r.get_refcnt() == 0;
|
||||
},
|
||||
[this, &key, &shard, &queued, &now](Replica& replica) {
|
||||
if (queued) return; // only need to pin one replica for offload
|
||||
auto result = PushOffloadingQueue(key, replica);
|
||||
if (result) {
|
||||
replica.inc_refcnt();
|
||||
shard->offloading_tasks.emplace(
|
||||
key, OffloadingTask{replica.id(), now});
|
||||
queued = true;
|
||||
}
|
||||
});
|
||||
|
||||
if (queued) {
|
||||
offload_queued_this_cycle++;
|
||||
offload_deferred_count++;
|
||||
// Any remaining MEMORY replicas with refcnt==0 are redundant copies
|
||||
// (data survives via the pinned replica → disk). Evict them now to
|
||||
// reclaim memory immediately rather than waiting another cycle.
|
||||
return metadata.size * evict_replicas(metadata);
|
||||
}
|
||||
|
||||
// PushOffloadingQueue failed. Default (data-preserving) behavior is to
|
||||
// skip this cycle — the outer eviction loop will retry after the
|
||||
// offload queue drains. Only force-evict when explicitly opted in, to
|
||||
// prevent silent data loss when the queue is unavailable.
|
||||
if (offload_force_evict_) {
|
||||
offload_push_failed_forced++;
|
||||
return metadata.size * evict_replicas(metadata);
|
||||
}
|
||||
return 0;
|
||||
};
|
||||
|
||||
// Randomly select a starting shard to avoid imbalance eviction between
|
||||
// shards. No need to use expensive random_device here.
|
||||
size_t start_idx = rand() % kNumShards;
|
||||
|
|
@ -3658,18 +3563,16 @@ void MasterService::BatchEvict(double evict_ratio_target,
|
|||
continue;
|
||||
}
|
||||
if (it->second.lease_timeout <= target_timeout) {
|
||||
// Evict this object (or defer for offload)
|
||||
uint64_t freed =
|
||||
try_evict_or_offload(it->first, it->second, shard);
|
||||
total_freed_size += freed;
|
||||
// Evict this object
|
||||
total_freed_size +=
|
||||
it->second.size *
|
||||
evict_replicas(it->second); // Erase memory replicas
|
||||
if (it->second.IsValid() == false) {
|
||||
it = shard->metadata.erase(it);
|
||||
} else {
|
||||
++it;
|
||||
}
|
||||
if (freed > 0) {
|
||||
shard_evicted_count++;
|
||||
}
|
||||
shard_evicted_count++;
|
||||
} else {
|
||||
// second pass candidates
|
||||
no_pin_objects.push_back(it->second.lease_timeout);
|
||||
|
|
@ -3721,18 +3624,17 @@ void MasterService::BatchEvict(double evict_ratio_target,
|
|||
it->second.lease_timeout <= target_timeout &&
|
||||
!it->second.IsSoftPinned(now) &&
|
||||
can_evict_replicas(it->second)) {
|
||||
// Evict this object (or defer for offload)
|
||||
uint64_t freed =
|
||||
try_evict_or_offload(it->first, it->second, shard);
|
||||
total_freed_size += freed;
|
||||
// Evict this object
|
||||
total_freed_size +=
|
||||
it->second.size *
|
||||
evict_replicas(
|
||||
it->second); // Erase memory replicas
|
||||
if (it->second.IsValid() == false) {
|
||||
it = shard->metadata.erase(it);
|
||||
} else {
|
||||
++it;
|
||||
}
|
||||
if (freed > 0) {
|
||||
evicted_count++;
|
||||
}
|
||||
evicted_count++;
|
||||
target_evict_num--;
|
||||
} else {
|
||||
++it;
|
||||
|
|
@ -3772,18 +3674,16 @@ void MasterService::BatchEvict(double evict_ratio_target,
|
|||
// and lease timeout less than or equal to target.
|
||||
if (!it->second.IsSoftPinned(now) ||
|
||||
it->second.lease_timeout <= soft_target_timeout) {
|
||||
// Evict this object (or defer for offload)
|
||||
uint64_t freed =
|
||||
try_evict_or_offload(it->first, it->second, shard);
|
||||
total_freed_size += freed;
|
||||
total_freed_size +=
|
||||
it->second.size *
|
||||
evict_replicas(
|
||||
it->second); // Erase memory replicas
|
||||
if (it->second.IsValid() == false) {
|
||||
it = shard->metadata.erase(it);
|
||||
} else {
|
||||
++it;
|
||||
}
|
||||
if (freed > 0) {
|
||||
evicted_count++;
|
||||
}
|
||||
evicted_count++;
|
||||
target_evict_num--;
|
||||
} else {
|
||||
++it;
|
||||
|
|
@ -3804,11 +3704,7 @@ void MasterService::BatchEvict(double evict_ratio_target,
|
|||
}
|
||||
}
|
||||
|
||||
if (evicted_count > 0 || released_discarded_cnt > 0 ||
|
||||
offload_deferred_count > 0) {
|
||||
// Offload-deferred counts as partial success: work was done (objects
|
||||
// queued for disk offload), so suppress re-triggering until the next
|
||||
// watermark breach or explicit need_eviction_ signal.
|
||||
if (evicted_count > 0 || released_discarded_cnt > 0) {
|
||||
need_eviction_ = false;
|
||||
MasterMetricManager::instance().inc_eviction_success(evicted_count,
|
||||
total_freed_size);
|
||||
|
|
@ -3820,27 +3716,7 @@ void MasterService::BatchEvict(double evict_ratio_target,
|
|||
MasterMetricManager::instance().inc_eviction_fail();
|
||||
}
|
||||
VLOG(1) << "action=evict_objects" << ", evicted_count=" << evicted_count
|
||||
<< ", offload_deferred=" << offload_deferred_count
|
||||
<< ", offload_cap_forced=" << offload_cap_forced_count
|
||||
<< ", offload_push_failed_forced=" << offload_push_failed_forced
|
||||
<< ", total_freed_size=" << total_freed_size;
|
||||
if (offload_on_evict_ && evicted_count == 0 && offload_deferred_count > 0) {
|
||||
LOG(WARNING) << "[EVICT] No memory freed this cycle; "
|
||||
<< offload_deferred_count
|
||||
<< " objects deferred for disk offload. "
|
||||
"Consider lowering eviction_high_watermark_ratio.";
|
||||
}
|
||||
if (offload_cap_forced_count > 0) {
|
||||
LOG(WARNING) << "[EVICT] Offload cap (" << offload_cap
|
||||
<< ") reached; force-evicted " << offload_cap_forced_count
|
||||
<< " object(s) without disk offload this cycle.";
|
||||
}
|
||||
if (offload_push_failed_forced > 0) {
|
||||
LOG(WARNING) << "[EVICT] PushOffloadingQueue failed for "
|
||||
<< offload_push_failed_forced
|
||||
<< " object(s); force-evicted without disk offload "
|
||||
"(offload_force_evict=true).";
|
||||
}
|
||||
}
|
||||
|
||||
void MasterService::ClientMonitorFunc() {
|
||||
|
|
|
|||
|
|
@ -91,10 +91,6 @@ int main(int argc, char *argv[]) {
|
|||
mooncake::ResourceTracker::getInstance();
|
||||
|
||||
gflags::ParseCommandLineFlags(&argc, &argv, true);
|
||||
if (!FLAGS_log_dir.empty()) {
|
||||
google::InitGoogleLogging(argv[0]);
|
||||
}
|
||||
|
||||
size_t global_segment_size = string_to_byte_size(FLAGS_global_segment_size);
|
||||
#ifdef USE_ASCEND_DIRECT
|
||||
// just set to true, does not affect GPU process.
|
||||
|
|
|
|||
|
|
@ -39,7 +39,6 @@ add_store_test(eviction_strategy_test eviction_strategy_test.cpp)
|
|||
add_store_test(master_service_test master_service_test.cpp)
|
||||
add_store_test(batch_remove_test batch_remove_test.cpp)
|
||||
add_store_test(master_service_ssd_test master_service_ssd_test.cpp)
|
||||
add_store_test(offload_on_evict_test offload_on_evict_test.cpp)
|
||||
add_store_test(master_service_ssd_test_for_snapshot
|
||||
ha/snapshot/master_service_ssd_test_for_snapshot.cpp)
|
||||
add_store_test(client_integration_test client_integration_test.cpp)
|
||||
|
|
|
|||
|
|
@ -1,388 +0,0 @@
|
|||
#include "master_service.h"
|
||||
|
||||
#include <glog/logging.h>
|
||||
#include <gtest/gtest.h>
|
||||
|
||||
#include <atomic>
|
||||
#include <chrono>
|
||||
#include <memory>
|
||||
#include <string>
|
||||
#include <thread>
|
||||
#include <vector>
|
||||
|
||||
#include "types.h"
|
||||
|
||||
namespace mooncake::test {
|
||||
|
||||
class OffloadOnEvictTest : public ::testing::Test {
|
||||
protected:
|
||||
void SetUp() override {
|
||||
google::InitGoogleLogging("OffloadOnEvictTest");
|
||||
FLAGS_logtostderr = true;
|
||||
}
|
||||
|
||||
void TearDown() override { google::ShutdownGoogleLogging(); }
|
||||
|
||||
static constexpr size_t kDefaultSegmentBase = 0x300000000;
|
||||
|
||||
Segment MakeSegment(std::string name, size_t base, size_t size) const {
|
||||
Segment segment;
|
||||
segment.id = generate_uuid();
|
||||
segment.name = std::move(name);
|
||||
segment.base = base;
|
||||
segment.size = size;
|
||||
segment.te_endpoint = segment.name;
|
||||
return segment;
|
||||
}
|
||||
|
||||
struct MountedSegmentContext {
|
||||
UUID segment_id;
|
||||
UUID client_id;
|
||||
};
|
||||
|
||||
MountedSegmentContext PrepareSegment(MasterService& service,
|
||||
std::string name, size_t base,
|
||||
size_t size) const {
|
||||
Segment segment = MakeSegment(std::move(name), base, size);
|
||||
UUID client_id = generate_uuid();
|
||||
auto mount_result = service.MountSegment(segment, client_id);
|
||||
EXPECT_TRUE(mount_result.has_value());
|
||||
return {.segment_id = segment.id, .client_id = client_id};
|
||||
}
|
||||
|
||||
// Put an object and complete it.
|
||||
void PutObject(MasterService& service, const UUID& client_id,
|
||||
const std::string& key, size_t size = 1024) {
|
||||
ReplicateConfig config;
|
||||
config.replica_num = 1;
|
||||
auto put_start = service.PutStart(client_id, key, size, config);
|
||||
ASSERT_TRUE(put_start.has_value()) << "PutStart failed for key=" << key;
|
||||
auto put_end = service.PutEnd(client_id, key, ReplicaType::MEMORY);
|
||||
ASSERT_TRUE(put_end.has_value()) << "PutEnd failed for key=" << key;
|
||||
}
|
||||
|
||||
// Drain the offload queue via OffloadObjectHeartbeat.
|
||||
std::unordered_map<std::string, int64_t> DrainOffloadQueue(
|
||||
MasterService& service, const UUID& client_id) {
|
||||
auto res = service.OffloadObjectHeartbeat(client_id, true);
|
||||
if (!res) {
|
||||
return {};
|
||||
}
|
||||
return std::move(res.value());
|
||||
}
|
||||
|
||||
template <typename Predicate>
|
||||
void WaitUntil(
|
||||
Predicate&& predicate,
|
||||
std::chrono::milliseconds timeout = std::chrono::milliseconds(4000),
|
||||
std::chrono::milliseconds interval =
|
||||
std::chrono::milliseconds(50)) const {
|
||||
const auto deadline = std::chrono::steady_clock::now() + timeout;
|
||||
while (std::chrono::steady_clock::now() < deadline) {
|
||||
if (predicate()) {
|
||||
return;
|
||||
}
|
||||
std::this_thread::sleep_for(interval);
|
||||
}
|
||||
EXPECT_TRUE(predicate());
|
||||
}
|
||||
|
||||
// Fill a segment until PutStart fails, triggering eviction.
|
||||
// Returns the number of successful puts.
|
||||
int FillSegmentUntilEviction(MasterService& service, const UUID& client_id,
|
||||
const std::string& key_prefix,
|
||||
size_t object_size, int max_puts) {
|
||||
int success_puts = 0;
|
||||
for (int i = 0; i < max_puts; ++i) {
|
||||
std::string key = key_prefix + std::to_string(i);
|
||||
ReplicateConfig config;
|
||||
config.replica_num = 1;
|
||||
auto result = service.PutStart(client_id, key, object_size, config);
|
||||
if (result.has_value()) {
|
||||
auto end = service.PutEnd(client_id, key, ReplicaType::MEMORY);
|
||||
EXPECT_TRUE(end.has_value());
|
||||
success_puts++;
|
||||
} else {
|
||||
// Wait for eviction to process
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(50));
|
||||
}
|
||||
}
|
||||
return success_puts;
|
||||
}
|
||||
};
|
||||
|
||||
// =============================================================================
|
||||
// Combo A: Default config (offload at PutEnd)
|
||||
// =============================================================================
|
||||
|
||||
TEST_F(OffloadOnEvictTest, ComboA_OffloadAtPutEnd) {
|
||||
MasterServiceConfig config;
|
||||
config.enable_offload = true;
|
||||
config.default_kv_lease_ttl = 2000;
|
||||
auto service = std::make_unique<MasterService>(config);
|
||||
|
||||
constexpr size_t seg_size = 1024 * 1024 * 16;
|
||||
auto ctx =
|
||||
PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
|
||||
|
||||
// Mount local disk segment with offloading ENABLED
|
||||
auto mount_ld = service->MountLocalDiskSegment(ctx.client_id, true);
|
||||
ASSERT_TRUE(mount_ld.has_value());
|
||||
|
||||
// Put objects
|
||||
PutObject(*service, ctx.client_id, "key_a1");
|
||||
PutObject(*service, ctx.client_id, "key_a2");
|
||||
PutObject(*service, ctx.client_id, "key_a3");
|
||||
|
||||
// Default mode: PutEnd pushes to offload queue immediately
|
||||
auto queued = DrainOffloadQueue(*service, ctx.client_id);
|
||||
EXPECT_EQ(queued.size(), 3u)
|
||||
<< "Default: all 3 objects should be in offload queue after PutEnd";
|
||||
EXPECT_TRUE(queued.count("key_a1"));
|
||||
EXPECT_TRUE(queued.count("key_a2"));
|
||||
EXPECT_TRUE(queued.count("key_a3"));
|
||||
|
||||
service->RemoveAll();
|
||||
}
|
||||
|
||||
TEST_F(OffloadOnEvictTest, ComboA_EvictionWorks) {
|
||||
// Regression: eviction still works in default mode
|
||||
const uint64_t kv_lease_ttl = 2000;
|
||||
MasterServiceConfig config;
|
||||
config.enable_offload = true;
|
||||
config.default_kv_lease_ttl = kv_lease_ttl;
|
||||
auto service = std::make_unique<MasterService>(config);
|
||||
|
||||
// Large segment: can hold ~16K objects of 15KB
|
||||
constexpr size_t seg_size = 1024 * 1024 * 16 * 15;
|
||||
constexpr size_t object_size = 1024 * 15;
|
||||
auto ctx =
|
||||
PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
|
||||
|
||||
// Put more objects than the segment can hold
|
||||
int success_puts = FillSegmentUntilEviction(
|
||||
*service, ctx.client_id, "evict_a_", object_size, 1024 * 16 + 50);
|
||||
EXPECT_GT(success_puts, 1024 * 16)
|
||||
<< "Default: eviction should allow more puts than capacity";
|
||||
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(kv_lease_ttl));
|
||||
service->RemoveAll();
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// Combo B: offload_on_evict=true (offload on evict, no force-evict)
|
||||
// =============================================================================
|
||||
|
||||
TEST_F(OffloadOnEvictTest, ComboB_PutEndSkipsOffloadQueue) {
|
||||
MasterServiceConfig config;
|
||||
config.enable_offload = true;
|
||||
config.offload_on_evict = true;
|
||||
config.default_kv_lease_ttl = 2000;
|
||||
auto service = std::make_unique<MasterService>(config);
|
||||
|
||||
constexpr size_t seg_size = 1024 * 1024 * 16;
|
||||
auto ctx =
|
||||
PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
|
||||
auto mount_ld = service->MountLocalDiskSegment(ctx.client_id, true);
|
||||
ASSERT_TRUE(mount_ld.has_value());
|
||||
|
||||
PutObject(*service, ctx.client_id, "key_b1");
|
||||
PutObject(*service, ctx.client_id, "key_b2");
|
||||
PutObject(*service, ctx.client_id, "key_b3");
|
||||
|
||||
// Offload-on-evict: PutEnd should NOT push to offload queue
|
||||
auto queued = DrainOffloadQueue(*service, ctx.client_id);
|
||||
EXPECT_EQ(queued.size(), 0u)
|
||||
<< "Offload-on-evict: queue should be empty after PutEnd";
|
||||
|
||||
service->RemoveAll();
|
||||
}
|
||||
|
||||
TEST_F(OffloadOnEvictTest, ComboB_EvictionTriggersOffload) {
|
||||
const uint64_t kv_lease_ttl = 2000;
|
||||
MasterServiceConfig config;
|
||||
config.enable_offload = true;
|
||||
config.offload_on_evict = true;
|
||||
config.default_kv_lease_ttl = kv_lease_ttl;
|
||||
auto service = std::make_unique<MasterService>(config);
|
||||
|
||||
constexpr size_t seg_size = 1024 * 1024 * 16 * 15;
|
||||
constexpr size_t object_size = 1024 * 15;
|
||||
auto ctx =
|
||||
PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
|
||||
auto mount_ld = service->MountLocalDiskSegment(ctx.client_id, true);
|
||||
ASSERT_TRUE(mount_ld.has_value());
|
||||
|
||||
// Fill segment to trigger eviction
|
||||
bool eviction_triggered = false;
|
||||
int success_puts = 0;
|
||||
for (int i = 0; i < 1024 * 16 + 50; ++i) {
|
||||
std::string key = "evict_b_" + std::to_string(i);
|
||||
ReplicateConfig config;
|
||||
config.replica_num = 1;
|
||||
auto result =
|
||||
service->PutStart(ctx.client_id, key, object_size, config);
|
||||
if (result.has_value()) {
|
||||
auto end = service->PutEnd(ctx.client_id, key, ReplicaType::MEMORY);
|
||||
ASSERT_TRUE(end.has_value());
|
||||
success_puts++;
|
||||
} else {
|
||||
eviction_triggered = true;
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(50));
|
||||
}
|
||||
}
|
||||
|
||||
EXPECT_TRUE(eviction_triggered)
|
||||
<< "Eviction should trigger when segment fills up";
|
||||
|
||||
// Offload-on-evict: eviction should push objects to offload queue
|
||||
auto queued = DrainOffloadQueue(*service, ctx.client_id);
|
||||
EXPECT_GT(queued.size(), 0u)
|
||||
<< "Offload-on-evict: eviction should push to offload queue";
|
||||
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(kv_lease_ttl));
|
||||
service->RemoveAll();
|
||||
}
|
||||
|
||||
TEST_F(OffloadOnEvictTest, ComboB_NoFallbackWithoutForceEvict) {
|
||||
// Without force_evict AND without a LocalDiskSegment, offload queue push
|
||||
// fails and eviction does NOT force-delete MEMORY (data-preserving).
|
||||
// The segment fills and subsequent puts fail — this is the safe default.
|
||||
const uint64_t kv_lease_ttl = 2000;
|
||||
MasterServiceConfig config;
|
||||
config.enable_offload = true;
|
||||
config.offload_on_evict = true;
|
||||
config.default_kv_lease_ttl = kv_lease_ttl;
|
||||
auto service = std::make_unique<MasterService>(config);
|
||||
|
||||
// NO local disk segment mounted — PushOffloadingQueue will fail
|
||||
constexpr size_t seg_size = 1024 * 1024 * 16 * 15;
|
||||
constexpr size_t object_size = 1024 * 15;
|
||||
auto ctx =
|
||||
PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
|
||||
|
||||
// Without force_evict, push failures mean DRAM cannot be freed,
|
||||
// so we can only put up to segment capacity (no overflow).
|
||||
int success_puts = FillSegmentUntilEviction(
|
||||
*service, ctx.client_id, "evict_b2_", object_size, 1024 * 16 + 50);
|
||||
EXPECT_LE(success_puts, 1024 * 16)
|
||||
<< "Without force_evict, segment should fill and stay full";
|
||||
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(kv_lease_ttl));
|
||||
service->RemoveAll();
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// Combo C: offload_on_evict=true + offload_force_evict=true
|
||||
// =============================================================================
|
||||
|
||||
TEST_F(OffloadOnEvictTest, ComboC_PutEndSkipsOffloadQueue) {
|
||||
MasterServiceConfig config;
|
||||
config.enable_offload = true;
|
||||
config.offload_on_evict = true;
|
||||
config.offload_force_evict = true;
|
||||
config.default_kv_lease_ttl = 2000;
|
||||
auto service = std::make_unique<MasterService>(config);
|
||||
|
||||
constexpr size_t seg_size = 1024 * 1024 * 16;
|
||||
auto ctx =
|
||||
PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
|
||||
auto mount_ld = service->MountLocalDiskSegment(ctx.client_id, true);
|
||||
ASSERT_TRUE(mount_ld.has_value());
|
||||
|
||||
PutObject(*service, ctx.client_id, "key_c1");
|
||||
PutObject(*service, ctx.client_id, "key_c2");
|
||||
|
||||
// Same as Combo B: PutEnd should skip offload queue
|
||||
auto queued = DrainOffloadQueue(*service, ctx.client_id);
|
||||
EXPECT_EQ(queued.size(), 0u)
|
||||
<< "Combo C: offload queue should be empty after PutEnd";
|
||||
|
||||
service->RemoveAll();
|
||||
}
|
||||
|
||||
TEST_F(OffloadOnEvictTest, ComboC_EvictionWithForceEvict) {
|
||||
const uint64_t kv_lease_ttl = 2000;
|
||||
MasterServiceConfig config;
|
||||
config.enable_offload = true;
|
||||
config.offload_on_evict = true;
|
||||
config.offload_force_evict = true;
|
||||
config.default_kv_lease_ttl = kv_lease_ttl;
|
||||
auto service = std::make_unique<MasterService>(config);
|
||||
|
||||
constexpr size_t seg_size = 1024 * 1024 * 16 * 15;
|
||||
constexpr size_t object_size = 1024 * 15;
|
||||
auto ctx =
|
||||
PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
|
||||
auto mount_ld = service->MountLocalDiskSegment(ctx.client_id, true);
|
||||
ASSERT_TRUE(mount_ld.has_value());
|
||||
|
||||
// With force-evict, eviction should work effectively.
|
||||
// Note: without a real FileStorage heartbeat, offloaded objects' refcnt
|
||||
// never decreases, so DRAM isn't fully freed beyond what direct eviction
|
||||
// allows. We verify eviction doesn't deadlock (can fill to capacity).
|
||||
int success_puts = FillSegmentUntilEviction(
|
||||
*service, ctx.client_id, "evict_c_", object_size, 1024 * 16 + 50);
|
||||
EXPECT_GE(success_puts, 1024 * 16)
|
||||
<< "Combo C: eviction should work with force-evict enabled";
|
||||
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(kv_lease_ttl));
|
||||
service->RemoveAll();
|
||||
}
|
||||
|
||||
// =============================================================================
|
||||
// Combo D: offload_force_evict=true only (should be no-op without on_evict)
|
||||
// =============================================================================
|
||||
|
||||
TEST_F(OffloadOnEvictTest, ComboD_ForceEvictAloneIsIgnored) {
|
||||
MasterServiceConfig config;
|
||||
config.enable_offload = true;
|
||||
config.offload_force_evict = true; // on_evict is false → force is ignored
|
||||
config.default_kv_lease_ttl = 2000;
|
||||
auto service = std::make_unique<MasterService>(config);
|
||||
|
||||
constexpr size_t seg_size = 1024 * 1024 * 16;
|
||||
auto ctx =
|
||||
PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
|
||||
auto mount_ld = service->MountLocalDiskSegment(ctx.client_id, true);
|
||||
ASSERT_TRUE(mount_ld.has_value());
|
||||
|
||||
// Should behave like Combo A (default: offload at PutEnd)
|
||||
PutObject(*service, ctx.client_id, "key_d1");
|
||||
PutObject(*service, ctx.client_id, "key_d2");
|
||||
|
||||
auto queued = DrainOffloadQueue(*service, ctx.client_id);
|
||||
EXPECT_EQ(queued.size(), 2u)
|
||||
<< "Combo D: FORCE_EVICT alone should not change default behavior";
|
||||
|
||||
service->RemoveAll();
|
||||
}
|
||||
|
||||
TEST_F(OffloadOnEvictTest, ComboD_EvictionWorks) {
|
||||
const uint64_t kv_lease_ttl = 2000;
|
||||
MasterServiceConfig config;
|
||||
config.enable_offload = true;
|
||||
config.offload_force_evict = true; // on_evict is false → force is ignored
|
||||
config.default_kv_lease_ttl = kv_lease_ttl;
|
||||
auto service = std::make_unique<MasterService>(config);
|
||||
|
||||
constexpr size_t seg_size = 1024 * 1024 * 16 * 15;
|
||||
constexpr size_t object_size = 1024 * 15;
|
||||
auto ctx =
|
||||
PrepareSegment(*service, "test_segment", kDefaultSegmentBase, seg_size);
|
||||
|
||||
int success_puts = FillSegmentUntilEviction(
|
||||
*service, ctx.client_id, "evict_d_", object_size, 1024 * 16 + 50);
|
||||
EXPECT_GT(success_puts, 1024 * 16)
|
||||
<< "Combo D: eviction should work normally";
|
||||
|
||||
std::this_thread::sleep_for(std::chrono::milliseconds(kv_lease_ttl));
|
||||
service->RemoveAll();
|
||||
}
|
||||
|
||||
} // namespace mooncake::test
|
||||
|
||||
int main(int argc, char** argv) {
|
||||
::testing::InitGoogleTest(&argc, argv);
|
||||
return RUN_ALL_TESTS();
|
||||
}
|
||||
|
|
@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
|||
|
||||
[project]
|
||||
name = "mooncake-transfer-engine"
|
||||
version = "0.3.10.post2"
|
||||
version = "0.3.10.post1"
|
||||
description = "Python binding of a Mooncake library using pybind11"
|
||||
authors = [
|
||||
{ name = "Mooncake Authors" }
|
||||
|
|
|
|||
|
|
@ -1,144 +0,0 @@
|
|||
"""Python-binding test for the offload-on-evict feature.
|
||||
|
||||
Unlike the default ``enable_ssd_offload`` path where PutEnd synchronously
|
||||
queues objects for offload, the offload-on-evict mode defers offload until
|
||||
DRAM eviction triggers. This test asserts the core behavioral difference:
|
||||
small workloads below the eviction watermark stay in DRAM only and do not
|
||||
produce LOCAL_DISK replicas.
|
||||
|
||||
The end-to-end overflow → evict → offload → read contract is covered by
|
||||
``test_ssd_offload_in_evict.py`` (legacy sync-offload mode) and the
|
||||
``verify_offload_on_eviction`` scenario in the kv-cache-offload-benchmark
|
||||
project, both of which exercise the same storage backend code paths as
|
||||
this feature.
|
||||
|
||||
Prerequisites:
|
||||
- mooncake_master running with master.json containing:
|
||||
``"enable_offload": true``
|
||||
``"offload_on_evict": true``
|
||||
- ``MOONCAKE_OFFLOAD_FILE_STORAGE_PATH=/tmp/...`` env var set on the
|
||||
client side so FileStorage uses the matching backend / path.
|
||||
"""
|
||||
|
||||
import unittest
|
||||
import os
|
||||
import time
|
||||
from mooncake.store import MooncakeDistributedStore
|
||||
|
||||
# Align with the master's default_kv_lease_ttl (override via env).
|
||||
DEFAULT_DEFAULT_KV_LEASE_TTL = 5000 # ms
|
||||
default_kv_lease_ttl = int(
|
||||
os.getenv("DEFAULT_KV_LEASE_TTL", DEFAULT_DEFAULT_KV_LEASE_TTL)
|
||||
)
|
||||
|
||||
SEGMENT_SIZE = int(os.getenv("SEGMENT_SIZE_BYTES", str(128 * 1024 * 1024)))
|
||||
LOCAL_BUFFER_SIZE = int(os.getenv("LOCAL_BUFFER_SIZE_BYTES", str(128 * 1024 * 1024)))
|
||||
|
||||
|
||||
def get_client(store):
|
||||
"""Initialize and setup the distributed store client."""
|
||||
protocol = os.getenv("PROTOCOL", "tcp")
|
||||
device_name = os.getenv("DEVICE_NAME", "")
|
||||
local_hostname = os.getenv("LOCAL_HOSTNAME", "localhost")
|
||||
metadata_server = os.getenv("MC_METADATA_SERVER", "P2PHANDSHAKE")
|
||||
master_server_address = os.getenv("MASTER_SERVER", "127.0.0.1:50051")
|
||||
|
||||
retcode = store.setup(
|
||||
local_hostname,
|
||||
metadata_server,
|
||||
SEGMENT_SIZE,
|
||||
LOCAL_BUFFER_SIZE,
|
||||
protocol,
|
||||
device_name,
|
||||
master_server_address,
|
||||
None, # engine
|
||||
True, # enable_ssd_offload — required for FileStorage / LOCAL_DISK
|
||||
)
|
||||
if retcode:
|
||||
raise RuntimeError(f"Failed to setup store client. Return code: {retcode}")
|
||||
|
||||
|
||||
class TestOffloadOnEviction(unittest.TestCase):
|
||||
"""Python-binding test for offload-on-evict behavioral difference."""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.store = MooncakeDistributedStore()
|
||||
get_client(cls.store)
|
||||
|
||||
def test_small_workload_no_disk_replica(self):
|
||||
"""Workload well below the eviction watermark must NOT produce
|
||||
LOCAL_DISK replicas.
|
||||
|
||||
Under the default ``enable_ssd_offload=true`` path, every PutEnd
|
||||
would queue the key for offload (heartbeat → disk). Under
|
||||
offload-on-evict, the offload queue stays empty until DRAM
|
||||
pressure triggers eviction. This asymmetry is what the feature
|
||||
is for — avoiding redundant disk I/O for hot data.
|
||||
|
||||
A zero disk-replica count thus doubles as a guard that the
|
||||
master is actually running with ``offload_on_evict: true`` in config.
|
||||
"""
|
||||
VALUE_SIZE = 1024 # 1 KB
|
||||
NUM_KEYS = 16 # 16 KB total, several orders of magnitude below
|
||||
# the 128 MB segment — eviction cannot trigger.
|
||||
|
||||
timestamp = int(time.time())
|
||||
keys = [f"ooe_small_{i}_{timestamp}" for i in range(NUM_KEYS)]
|
||||
reference = {}
|
||||
|
||||
try:
|
||||
for key in keys:
|
||||
value = os.urandom(VALUE_SIZE)
|
||||
retcode = self.store.put(key, value)
|
||||
self.assertEqual(
|
||||
retcode,
|
||||
0,
|
||||
f"Put failed for {key}: retcode={retcode}",
|
||||
)
|
||||
reference[key] = value
|
||||
|
||||
# Wait past one heartbeat interval so any async offload would
|
||||
# have had a chance to complete.
|
||||
time.sleep(3)
|
||||
|
||||
# Query replicas — they must all be MEMORY only.
|
||||
descs = self.store.batch_get_replica_desc(keys)
|
||||
disk_replicas = 0
|
||||
for key in keys:
|
||||
infos = descs.get(key) if isinstance(descs, dict) else None
|
||||
if infos is None:
|
||||
continue
|
||||
if not isinstance(infos, (list, tuple)):
|
||||
infos = [infos]
|
||||
for info in infos:
|
||||
type_str = str(getattr(info, "type", ""))
|
||||
if type_str and "MEMORY" not in type_str:
|
||||
disk_replicas += 1
|
||||
|
||||
self.assertEqual(
|
||||
disk_replicas,
|
||||
0,
|
||||
f"Offload-on-evict should keep small workloads in DRAM "
|
||||
f"only, but found {disk_replicas} LOCAL_DISK replicas. "
|
||||
f"Is offload_on_evict=true set in master config?",
|
||||
)
|
||||
|
||||
# Sanity: every key is readable and bit-exact (served from DRAM).
|
||||
for key, expected in reference.items():
|
||||
self.assertEqual(
|
||||
self.store.get(key), expected, f"Bit-exact mismatch for {key}"
|
||||
)
|
||||
finally:
|
||||
# Cleanup — remove all keys written by this test regardless of
|
||||
# success or failure so subsequent tests start with clean DRAM.
|
||||
for key in keys:
|
||||
try:
|
||||
self.store.remove(key)
|
||||
except Exception:
|
||||
pass
|
||||
time.sleep(default_kv_lease_ttl / 1000 + 0.5)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
@ -170,8 +170,11 @@ rm -rf ${OUTPUT_DIR}/
|
|||
mkdir -p ${OUTPUT_DIR}
|
||||
|
||||
echo "Installing required build packages"
|
||||
if command -v pip &>/dev/null; then
|
||||
if command -v python${PYTHON_VERSION} &>/dev/null; then
|
||||
python${PYTHON_VERSION} -m pip install --upgrade pip build setuptools wheel auditwheel
|
||||
elif command -v pip &>/dev/null; then
|
||||
pip install --upgrade pip
|
||||
pip install build setuptools wheel auditwheel
|
||||
elif command -v uv &>/dev/null; then
|
||||
uv pip install --upgrade pip
|
||||
uv pip install build setuptools wheel auditwheel
|
||||
|
|
|
|||
Loading…
Reference in New Issue