Merge pull request #9846 from blackpiglet/jxun/cbt_new

Implement the CBT service.
2026-05-31 11:26:22 +00:00 · 2026-05-27 15:08:46 +08:00
parent a606610e02 8bca779270
commit 2679db6b42
8 changed files with 497 additions and 32 deletions
--- a/changelogs/unreleased/9846-blackpiglet
+++ b/changelogs/unreleased/9846-blackpiglet
@@ -0,0 +1 @@
+Implement the CBT service.
--- a/go.mod
+++ b/go.mod
@@ -27,6 +27,7 @@ require (
 	github.com/hashicorp/go-plugin v1.7.0
 	github.com/joho/godotenv v1.3.0
 	github.com/kopia/kopia v0.16.0
+	github.com/kubernetes-csi/external-snapshot-metadata v1.0.0
 	github.com/kubernetes-csi/external-snapshotter/client/v8 v8.4.0
 	github.com/onsi/ginkgo/v2 v2.28.3
 	github.com/onsi/gomega v1.40.0
@@ -111,14 +112,25 @@ require (
 	github.com/go-logr/logr v1.4.3 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/go-ole/go-ole v1.3.0 // 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/go-openapi/jsonpointer v0.22.5 // indirect
+	github.com/go-openapi/jsonreference v0.21.5 // indirect
+	github.com/go-openapi/swag v0.25.5 // indirect
+	github.com/go-openapi/swag/cmdutils v0.25.5 // indirect
+	github.com/go-openapi/swag/conv v0.25.5 // indirect
+	github.com/go-openapi/swag/fileutils v0.25.5 // indirect
+	github.com/go-openapi/swag/jsonname v0.25.5 // indirect
+	github.com/go-openapi/swag/jsonutils v0.25.5 // indirect
+	github.com/go-openapi/swag/loading v0.25.5 // indirect
+	github.com/go-openapi/swag/mangling v0.25.5 // indirect
+	github.com/go-openapi/swag/netutils v0.25.5 // indirect
+	github.com/go-openapi/swag/stringutils v0.25.5 // indirect
+	github.com/go-openapi/swag/typeutils v0.25.5 // indirect
+	github.com/go-openapi/swag/yamlutils v0.25.5 // indirect
 	github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
 	github.com/gofrs/flock v0.13.0 // indirect
 	github.com/golang-jwt/jwt/v5 v5.3.1 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
-	github.com/google/gnostic-models v0.7.0 // indirect
+	github.com/google/gnostic-models v0.7.1 // indirect
 	github.com/google/pprof v0.0.0-20260402051712-545e8a4df936 // indirect
 	github.com/google/s2a-go v0.1.9 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.3.15 // indirect
@@ -128,16 +140,15 @@ require (
 	github.com/hashicorp/yamux v0.1.2 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jmespath/go-jmespath v0.4.0 // indirect
-	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
 	github.com/klauspost/compress v1.18.6 // indirect
 	github.com/klauspost/cpuid/v2 v2.3.0 // indirect
 	github.com/klauspost/crc32 v1.3.0 // indirect
 	github.com/klauspost/pgzip v1.2.6 // indirect
 	github.com/klauspost/reedsolomon v1.14.0 // indirect
+	github.com/kubernetes-csi/external-snapshot-metadata/client v1.0.0 // indirect
 	github.com/kylelemons/godebug v1.1.0 // indirect
 	github.com/liggitt/tabwriter v0.0.0-20181228230101-89fcab3d43de // indirect
-	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mattn/go-colorable v0.1.14 // indirect
 	github.com/mattn/go-isatty v0.0.21 // indirect
 	github.com/minio/crc64nvme v1.1.1 // indirect
@@ -158,7 +169,7 @@ require (
 	github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/prometheus/common v0.67.5 // indirect
-	github.com/prometheus/procfs v0.19.2 // indirect
+	github.com/prometheus/procfs v0.20.1 // indirect
 	github.com/rs/xid v1.6.0 // indirect
 	github.com/spiffe/go-spiffe/v2 v2.6.0 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
@@ -178,7 +189,7 @@ require (
 	go.opentelemetry.io/otel/trace v1.43.0 // indirect
 	go.starlark.net v0.0.0-20230525235612-a134d8f9ddca // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	go.yaml.in/yaml/v2 v2.4.3 // indirect
+	go.yaml.in/yaml/v2 v2.4.4 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
 	golang.org/x/crypto v0.50.0 // indirect
 	golang.org/x/exp v0.0.0-20260410095643-746e56fc9e2f // indirect
--- a/go.sum
+++ b/go.sum
@@ -174,6 +174,8 @@ github.com/bits-and-blooms/bitset v1.12.0 h1:U/q1fAF7xXRhFCrhROzIfffYnu+dlS38vCZ
 github.com/bits-and-blooms/bitset v1.12.0/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8=
 github.com/bketelsen/crypt v0.0.3-0.20200106085610-5cbc8cc4026c/go.mod h1:MKsuJmJgSg28kpZDP6UIiPt0e0Oz0kqKNGyRaWEPv84=
 github.com/bketelsen/crypt v0.0.4/go.mod h1:aI6NrJ0pMGgvZKL1iVgXLnfIFJtfV+bKCoqOes/6LfM=
+github.com/blang/semver/v4 v4.0.0 h1:1PFHFE6yCCTv8C1TeyNNarDzntLi7wMI5i/pzqYIsAM=
+github.com/blang/semver/v4 v4.0.0/go.mod h1:IbckMUScFkM3pff0VJDNKRiT6TG/YpiHIM2yvyW5YoQ=
 github.com/bombsimon/logrusr/v3 v3.0.0 h1:tcAoLfuAhKP9npBxWzSdpsvKPQt1XV02nSf2lZA82TQ=
 github.com/bombsimon/logrusr/v3 v3.0.0/go.mod h1:PksPPgSFEL2I52pla2glgCyyd2OqOHAnFF5E+g8Ixco=
 github.com/bufbuild/protocompile v0.14.1 h1:iA73zAf/fyljNjQKwYzUHD6AD4R8KMasmwa/FBatYVw=
@@ -193,6 +195,8 @@ github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnht
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2 h1:aBangftG7EVZoUb69Os8IaYg++6uMOdKK83QtkkvJik=
 github.com/cncf/xds/go v0.0.0-20260202195803-dba9d589def2/go.mod h1:qwXFYgsP6T7XnJtbKlf1HP8AjxZZyzxMmc+Lq5GjlU4=
+github.com/container-storage-interface/spec v1.12.0 h1:zrFOEqpR5AghNaaDG4qyedwPBqU2fU0dWjLQMP/azK0=
+github.com/container-storage-interface/spec v1.12.0/go.mod h1:txsm+MA2B2WDa5kW69jNbqPnvTtfvZma7T/zsAZ9qX8=
 github.com/coreos/bbolt v1.3.2/go.mod h1:iRUV2dpdMOn7Bo10OQBFzIJO9kkE559Wcmn+qkEiiKk=
 github.com/coreos/etcd v3.3.10+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
 github.com/coreos/etcd v3.3.13+incompatible/go.mod h1:uF7uidLiAD3TWHmW31ZFd/JWoc32PjwdhPthX9715RE=
@@ -289,18 +293,44 @@ github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE=
 github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78=
 github.com/go-openapi/jsonpointer v0.19.3/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
 github.com/go-openapi/jsonpointer v0.19.5/go.mod h1:Pl9vOtqEWErmShwVjC8pYs9cog34VGT37dQOVbmoatg=
-github.com/go-openapi/jsonpointer v0.19.6/go.mod h1:osyAmYz/mB/C3I+WsTTSgw1ONzaLJoLCyoi6/zppojs=
-github.com/go-openapi/jsonpointer v0.21.0 h1:YgdVicSA9vH5RiHs9TZW5oyafXZFc6+2Vc1rr/O9oNQ=
-github.com/go-openapi/jsonpointer v0.21.0/go.mod h1:IUyH9l/+uyhIYQ/PXVA41Rexl+kOkAPDdXEYns6fzUY=
+github.com/go-openapi/jsonpointer v0.22.5 h1:8on/0Yp4uTb9f4XvTrM2+1CPrV05QPZXu+rvu2o9jcA=
+github.com/go-openapi/jsonpointer v0.22.5/go.mod h1:gyUR3sCvGSWchA2sUBJGluYMbe1zazrYWIkWPjjMUY0=
 github.com/go-openapi/jsonreference v0.19.3/go.mod h1:rjx6GuL8TTa9VaixXglHmQmIL98+wF9xc8zWvFonSJ8=
 github.com/go-openapi/jsonreference v0.19.5/go.mod h1:RdybgQwPxbL4UEjuAruzK1x3nE69AqPYEJeo/TWfEeg=
-github.com/go-openapi/jsonreference v0.20.2 h1:3sVjiK66+uXK/6oQ8xgcRKcFgQ5KXa2KvnJRumpMGbE=
-github.com/go-openapi/jsonreference v0.20.2/go.mod h1:Bl1zwGIM8/wsvqjsOQLJ/SH+En5Ap4rVB5KVcIDZG2k=
+github.com/go-openapi/jsonreference v0.21.5 h1:6uCGVXU/aNF13AQNggxfysJ+5ZcU4nEAe+pJyVWRdiE=
+github.com/go-openapi/jsonreference v0.21.5/go.mod h1:u25Bw85sX4E2jzFodh1FOKMTZLcfifd1Q+iKKOUxExw=
 github.com/go-openapi/swag v0.19.5/go.mod h1:POnQmlKehdgb5mhVOsnJFsivZCEZ/vjK9gh66Z9tfKk=
 github.com/go-openapi/swag v0.19.14/go.mod h1:QYRuS/SOXUCsnplDa677K7+DxSOj6IPNl/eQntq43wQ=
-github.com/go-openapi/swag v0.22.3/go.mod h1:UzaqsxGiab7freDnrUUra0MwWfN/q7tE4j+VcZ0yl14=
-github.com/go-openapi/swag v0.23.0 h1:vsEVJDUo2hPJ2tu0/Xc+4noaxyEffXNIs3cOULZ+GrE=
-github.com/go-openapi/swag v0.23.0/go.mod h1:esZ8ITTYEsH1V2trKHjAN8Ai7xHb8RV+YSZ577vPjgQ=
+github.com/go-openapi/swag v0.25.5 h1:pNkwbUEeGwMtcgxDr+2GBPAk4kT+kJ+AaB+TMKAg+TU=
+github.com/go-openapi/swag v0.25.5/go.mod h1:B3RT6l8q7X803JRxa2e59tHOiZlX1t8viplOcs9CwTA=
+github.com/go-openapi/swag/cmdutils v0.25.5 h1:yh5hHrpgsw4NwM9KAEtaDTXILYzdXh/I8Whhx9hKj7c=
+github.com/go-openapi/swag/cmdutils v0.25.5/go.mod h1:pdae/AFo6WxLl5L0rq87eRzVPm/XRHM3MoYgRMvG4A0=
+github.com/go-openapi/swag/conv v0.25.5 h1:wAXBYEXJjoKwE5+vc9YHhpQOFj2JYBMF2DUi+tGu97g=
+github.com/go-openapi/swag/conv v0.25.5/go.mod h1:CuJ1eWvh1c4ORKx7unQnFGyvBbNlRKbnRyAvDvzWA4k=
+github.com/go-openapi/swag/fileutils v0.25.5 h1:B6JTdOcs2c0dBIs9HnkyTW+5gC+8NIhVBUwERkFhMWk=
+github.com/go-openapi/swag/fileutils v0.25.5/go.mod h1:V3cT9UdMQIaH4WiTrUc9EPtVA4txS0TOmRURmhGF4kc=
+github.com/go-openapi/swag/jsonname v0.25.5 h1:8p150i44rv/Drip4vWI3kGi9+4W9TdI3US3uUYSFhSo=
+github.com/go-openapi/swag/jsonname v0.25.5/go.mod h1:jNqqikyiAK56uS7n8sLkdaNY/uq6+D2m2LANat09pKU=
+github.com/go-openapi/swag/jsonutils v0.25.5 h1:XUZF8awQr75MXeC+/iaw5usY/iM7nXPDwdG3Jbl9vYo=
+github.com/go-openapi/swag/jsonutils v0.25.5/go.mod h1:48FXUaz8YsDAA9s5AnaUvAmry1UcLcNVWUjY42XkrN4=
+github.com/go-openapi/swag/jsonutils/fixtures_test v0.25.5 h1:SX6sE4FrGb4sEnnxbFL/25yZBb5Hcg1inLeErd86Y1U=
+github.com/go-openapi/swag/jsonutils/fixtures_test v0.25.5/go.mod h1:/2KvOTrKWjVA5Xli3DZWdMCZDzz3uV/T7bXwrKWPquo=
+github.com/go-openapi/swag/loading v0.25.5 h1:odQ/umlIZ1ZVRteI6ckSrvP6e2w9UTF5qgNdemJHjuU=
+github.com/go-openapi/swag/loading v0.25.5/go.mod h1:I8A8RaaQ4DApxhPSWLNYWh9NvmX2YKMoB9nwvv6oW6g=
+github.com/go-openapi/swag/mangling v0.25.5 h1:hyrnvbQRS7vKePQPHHDso+k6CGn5ZBs5232UqWZmJZw=
+github.com/go-openapi/swag/mangling v0.25.5/go.mod h1:6hadXM/o312N/h98RwByLg088U61TPGiltQn71Iw0NY=
+github.com/go-openapi/swag/netutils v0.25.5 h1:LZq2Xc2QI8+7838elRAaPCeqJnHODfSyOa7ZGfxDKlU=
+github.com/go-openapi/swag/netutils v0.25.5/go.mod h1:lHbtmj4m57APG/8H7ZcMMSWzNqIQcu0RFiXrPUara14=
+github.com/go-openapi/swag/stringutils v0.25.5 h1:NVkoDOA8YBgtAR/zvCx5rhJKtZF3IzXcDdwOsYzrB6M=
+github.com/go-openapi/swag/stringutils v0.25.5/go.mod h1:PKK8EZdu4QJq8iezt17HM8RXnLAzY7gW0O1KKarrZII=
+github.com/go-openapi/swag/typeutils v0.25.5 h1:EFJ+PCga2HfHGdo8s8VJXEVbeXRCYwzzr9u4rJk7L7E=
+github.com/go-openapi/swag/typeutils v0.25.5/go.mod h1:itmFmScAYE1bSD8C4rS0W+0InZUBrB2xSPbWt6DLGuc=
+github.com/go-openapi/swag/yamlutils v0.25.5 h1:kASCIS+oIeoc55j28T4o8KwlV2S4ZLPT6G0iq2SSbVQ=
+github.com/go-openapi/swag/yamlutils v0.25.5/go.mod h1:Gek1/SjjfbYvM+Iq4QGwa/2lEXde9n2j4a3wI3pNuOQ=
+github.com/go-openapi/testify/enable/yaml/v2 v2.4.0 h1:7SgOMTvJkM8yWrQlU8Jm18VeDPuAvB/xWrdxFJkoFag=
+github.com/go-openapi/testify/enable/yaml/v2 v2.4.0/go.mod h1:14iV8jyyQlinc9StD7w1xVPW3CO3q1Gj04Jy//Kw4VM=
+github.com/go-openapi/testify/v2 v2.4.0 h1:8nsPrHVCWkQ4p8h1EsRVymA2XABB4OT40gcvAu+voFM=
+github.com/go-openapi/testify/v2 v2.4.0/go.mod h1:HCPmvFFnheKK2BuwSA0TbbdxJ3I16pjwMkYkP4Ywn54=
 github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY=
 github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI=
 github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8=
@@ -332,6 +362,8 @@ github.com/golang/mock v1.4.1/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt
 github.com/golang/mock v1.4.3/go.mod h1:UOMv5ysSaYNkG+OFQykRIcU/QvvxJf3p21QfJ2Bt3cw=
 github.com/golang/mock v1.4.4/go.mod h1:l3mdAwkq5BuhzHwde/uurv3sEJeZMXNpwsxVWU71h+4=
 github.com/golang/mock v1.5.0/go.mod h1:CWnOUgYIOo4TcNZ0wHX3YZCqsaM1I1Jvs6v3mP3KVu8=
+github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc=
+github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
 github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
@@ -354,8 +386,8 @@ github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6
 github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
 github.com/google/btree v1.0.1/go.mod h1:xXMiIv4Fb/0kKde4SpL7qlzvu5cMJDRkFDxJfI9uaxA=
-github.com/google/gnostic-models v0.7.0 h1:qwTtogB15McXDaNqTZdzPJRHvaVJlAl+HVQnLmJEJxo=
-github.com/google/gnostic-models v0.7.0/go.mod h1:whL5G0m6dmc5cPxKc5bdKdEN3UjI7OUGxBlw57miDrQ=
+github.com/google/gnostic-models v0.7.1 h1:SisTfuFKJSKM5CPZkffwi6coztzzeYUhc3v4yxLWH8c=
+github.com/google/gnostic-models v0.7.1/go.mod h1:whL5G0m6dmc5cPxKc5bdKdEN3UjI7OUGxBlw57miDrQ=
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
 github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
 github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU=
@@ -470,7 +502,6 @@ github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfC
 github.com/joho/godotenv v1.3.0 h1:Zjp+RcGpHhGlrMbJzXTrZZPrWj+1vfm90La1wgB6Bhc=
 github.com/joho/godotenv v1.3.0/go.mod h1:7hK45KPybAkOC6peb+G5yklZfMxEjkZhHbwpqxOKXbg=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
-github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY=
 github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y=
 github.com/joshdk/go-junit v1.0.0 h1:S86cUKIdwBHWwA6xCmFlf3RTLfVXYQfvanM5Uh+K6GE=
 github.com/joshdk/go-junit v1.0.0/go.mod h1:TiiV0PqkaNfFXjEiyjWM3XXrhVyCa1K4Zfga6W52ung=
@@ -505,13 +536,20 @@ github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
 github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
-github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/kubernetes-csi/csi-lib-utils v0.23.2 h1:+x9W4RRyuRnJcTiSHKEFpl0cYUeLUqshM5ioPeYPRXw=
+github.com/kubernetes-csi/csi-lib-utils v0.23.2/go.mod h1:aIcqnC6EyesZpe7kX5PxHUZePw1tKrYFKwg7RaqlPh8=
+github.com/kubernetes-csi/csi-test/v5 v5.4.0 h1:u5DgYNIreSNO2+u4Nq2Wpl+bbakRSjNyxZHmDTAqnYA=
+github.com/kubernetes-csi/csi-test/v5 v5.4.0/go.mod h1:anAJKFUb/SdHhIHECgSKxC5LSiLzib+1I6mrWF5Hve8=
+github.com/kubernetes-csi/external-snapshot-metadata v1.0.0 h1:0UaIl/+A+wgwU8/vsh8XmgOvr4eBQMpq42CLQud3rRg=
+github.com/kubernetes-csi/external-snapshot-metadata v1.0.0/go.mod h1:kMm9tr8gdf0Dtr1JviPgi4rMy7A4cWFV7ML+0xF0bE0=
+github.com/kubernetes-csi/external-snapshot-metadata/client v1.0.0 h1:yRSDNeTNeZq7lSuDKXMv29ejtIRDgmJHCq36EjLQxLU=
+github.com/kubernetes-csi/external-snapshot-metadata/client v1.0.0/go.mod h1:Nufr1PWalLkuIyjS4Zr+ugyYEbK7MzbEUtZQUPeAJF0=
 github.com/kubernetes-csi/external-snapshotter/client/v8 v8.4.0 h1:bMqrb3UHgHbP+PW9VwiejfDJU1R0PpXVZNMdeH8WYKI=
 github.com/kubernetes-csi/external-snapshotter/client/v8 v8.4.0/go.mod h1:E3vdYxHj2C2q6qo8/Da4g7P+IcwqRZyy3gJBzYybV9Y=
 github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
@@ -525,8 +563,6 @@ github.com/mailru/easyjson v0.0.0-20190614124828-94de47d64c63/go.mod h1:C1wdFJiN
 github.com/mailru/easyjson v0.0.0-20190626092158-b2ccc519800e/go.mod h1:C1wdFJiN94OJF2b5HbByQZoLdCWB1Yqtg26g4irojpc=
 github.com/mailru/easyjson v0.7.0/go.mod h1:KAzv3t3aY1NaHWoQz1+4F1ccyAH66Jk7yos7ldAVICs=
 github.com/mailru/easyjson v0.7.6/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
-github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0=
-github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc=
 github.com/maruel/natural v1.1.1 h1:Hja7XhhmvEFhcByqDoHz9QZbkWey+COd9xWfCfn1ioo=
 github.com/maruel/natural v1.1.1/go.mod h1:v+Rfd79xlw1AgVBjbO0BEQmptqb5HvL/k9GRHB7ZKEg=
 github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU=
@@ -641,8 +677,8 @@ github.com/prometheus/common v0.67.5 h1:pIgK94WWlQt1WLwAC5j2ynLaBRDiinoAb86HZHTU
 github.com/prometheus/common v0.67.5/go.mod h1:SjE/0MzDEEAyrdr5Gqc6G+sXI67maCxzaT3A2+HqjUw=
 github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk=
 github.com/prometheus/procfs v0.0.0-20190507164030-5867b95ac084/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA=
-github.com/prometheus/procfs v0.19.2 h1:zUMhqEW66Ex7OXIiDkll3tl9a1ZdilUOd/F6ZXw4Vws=
-github.com/prometheus/procfs v0.19.2/go.mod h1:M0aotyiemPhBCM0z5w87kL22CxfcH05ZpYlu+b4J7mw=
+github.com/prometheus/procfs v0.20.1 h1:XwbrGOIplXW/AU3YhIhLODXMJYyC1isLFfYCsTEycfc=
+github.com/prometheus/procfs v0.20.1/go.mod h1:o9EMBZGRyvDrSPH1RqdxhojkuXstoe4UlK79eF5TGGo=
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
 github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
 github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
@@ -696,8 +732,6 @@ github.com/stoewer/go-strcase v1.2.0/go.mod h1:IBiWB2sKIp3wVVQ3Y035++gc+knqhUQag
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoHMkEqE=
-github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
-github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
 github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
 github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
 github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
@@ -706,10 +740,7 @@ github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81P
 github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
-github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.2/go.mod h1:R6va5+xMeoiuVRoj+gSkQ7d3FALtqAAGI1FQKckRals=
-github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
-github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/subosito/gotenv v1.2.0/go.mod h1:N0PQaV/YGNqwC0u51sEeR/aUtSLEXKX9iv69rRypqCw=
@@ -798,8 +829,8 @@ go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
 go.uber.org/zap v1.17.0/go.mod h1:MXVU+bhUf/A7Xi2HNOnopQOrmycQ5Ih87HtOu4q5SSo=
 go.uber.org/zap v1.28.0 h1:IZzaP1Fv73/T/pBMLk4VutPl36uNC+OSUh3JLG3FIjo=
 go.uber.org/zap v1.28.0/go.mod h1:rDLpOi171uODNm/mxFcuYWxDsqWSAVkFdX4XojSKg/Q=
-go.yaml.in/yaml/v2 v2.4.3 h1:6gvOSjQoTB3vt1l+CU+tSyi/HOjfOjRLJ4YwYZGwRO0=
-go.yaml.in/yaml/v2 v2.4.3/go.mod h1:zSxWcmIDjOzPXpjlTTbAsKokqkDNAVtZO0WOMiT90s8=
+go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ=
+go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ=
 go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc=
 go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
@@ -1253,6 +1284,8 @@ k8s.io/cli-runtime v0.36.0/go.mod h1:KObkknK9Ro5LYX+1RdiKc7C8CvGg4aX+V/Zv+E8WPHA
 k8s.io/client-go v0.22.2/go.mod h1:sAlhrkVDf50ZHx6z4K0S40wISNTarf1r800F+RlCF6U=
 k8s.io/client-go v0.36.0 h1:pOYi7C4RHChYjMiHpZSpSbIM6ZxVbRXBy7CuiIwqA3c=
 k8s.io/client-go v0.36.0/go.mod h1:ZKKcpwF0aLYfkHFCjillCKaTK/yBkEDHTDXCFY6AS9Y=
+k8s.io/component-base v0.36.0 h1:hFjEktssxiJhrK1zfybkH4kJOi8iZuF+mIDCqS5+jRo=
+k8s.io/component-base v0.36.0/go.mod h1:JZvIfcNHk+uck+8LhJzhSBtydWXaZNQwX2OdL+Mnwsk=
 k8s.io/gengo v0.0.0-20200413195148-3a45101e95ac/go.mod h1:ezvh/TsK7cY6rbqRK0oQQ8IAqLxYwwyPxAX1Pzy0ii0=
 k8s.io/klog/v2 v2.0.0/go.mod h1:PBfzABfn139FHAV07az/IF9Wp1bkk3vpT2XSJ76fSDE=
 k8s.io/klog/v2 v2.9.0/go.mod h1:hy9LJ/NvuK+iVyP4Ehqva4HxZG/oXyIS3n3Jmire4Ec=
--- a/pkg/cbtservice/csi_service_impl.go
+++ b/pkg/cbtservice/csi_service_impl.go
@@ -0,0 +1,130 @@
+/*
+Copyright The Velero Contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cbtservice
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/kubernetes-csi/external-snapshot-metadata/pkg/iterator"
+	"github.com/sirupsen/logrus"
+	"k8s.io/client-go/rest"
+)
+
+var (
+	buildClients        = iterator.BuildClients
+	getSnapshotMetadata = iterator.GetSnapshotMetadata
+)
+
+type emitterImpl struct {
+	logger         logrus.FieldLogger
+	recordCallBack func([]Range) error
+}
+
+func (e *emitterImpl) SnapshotMetadataIteratorRecord(recordNumber int, metadata iterator.IteratorMetadata) error {
+	for _, b := range metadata.BlockMetadata {
+		// Offset and size should not be negative, if they are, it indicates some error in the metadata and we should return error to stop the iteration.
+		if b.ByteOffset < 0 || b.SizeBytes < 0 {
+			return fmt.Errorf("invalid CBT metadata: offset: %v, size %v", b.ByteOffset, b.SizeBytes)
+		}
+
+		e.logger.Debugf("recording metadata for record number %d, offset: %v, size: %v",
+			recordNumber, b.ByteOffset, b.SizeBytes)
+
+		if err := e.recordCallBack([]Range{{Offset: uint64(b.ByteOffset), Length: uint64(b.SizeBytes)}}); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (e *emitterImpl) SnapshotMetadataIteratorDone(numberRecords int) error {
+	e.logger.Infof("finished iterating snapshot metadata, total number of records: %d", numberRecords)
+
+	return nil
+}
+
+type ServiceImpl struct {
+	logger       logrus.FieldLogger
+	vsNamespace  string
+	SAName       string
+	clientConfig *rest.Config
+}
+
+func NewService(
+	logger logrus.FieldLogger,
+	vsNamespace,
+	saName string,
+	clientConfig *rest.Config,
+) Service {
+	return &ServiceImpl{
+		logger:       logger,
+		vsNamespace:  vsNamespace,
+		SAName:       saName,
+		clientConfig: clientConfig,
+	}
+}
+
+func (s *ServiceImpl) GetAllocatedBlocks(ctx context.Context, snapshot string, record func([]Range) error) error {
+	clients, err := buildClients(s.clientConfig)
+	if err != nil {
+		return err
+	}
+
+	args := iterator.Args{
+		SnapshotName: snapshot,
+		Emitter: &emitterImpl{
+			logger:         s.logger,
+			recordCallBack: record,
+		},
+
+		Clients:         clients,
+		Namespace:       s.vsNamespace, // DataUpload is created in the same namespace as Velero server. vsNamespace is the namespace of the Velero server.
+		SANamespace:     s.vsNamespace, // The SA is created in the same namespace as Velero server. vsNamespace is the namespace of Velero	server.
+		SAName:          s.SAName,
+		TokenExpirySecs: iterator.DefaultTokenExpirySeconds,
+		MaxResults:      0, // If 0 then the CSI driver decides the value.
+	}
+
+	return getSnapshotMetadata(ctx, args)
+}
+
+func (s *ServiceImpl) GetChangedBlocks(ctx context.Context, snapshot string, changeID string, record func([]Range) error) error {
+	clients, err := buildClients(s.clientConfig)
+	if err != nil {
+		return err
+	}
+
+	args := iterator.Args{
+		SnapshotName:     snapshot,
+		PrevSnapshotName: changeID,
+		Emitter: &emitterImpl{
+			logger:         s.logger,
+			recordCallBack: record,
+		},
+
+		Clients:         clients,
+		Namespace:       s.vsNamespace,
+		SANamespace:     s.vsNamespace,
+		SAName:          s.SAName,
+		TokenExpirySecs: iterator.DefaultTokenExpirySeconds,
+		MaxResults:      0, // If 0 then the CSI driver decides the value.
+	}
+
+	return getSnapshotMetadata(ctx, args)
+}
--- a/pkg/cbtservice/csi_service_impl_test.go
+++ b/pkg/cbtservice/csi_service_impl_test.go
@@ -0,0 +1,258 @@
+/*
+Copyright The Velero Contributors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package cbtservice
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/kubernetes-csi/external-snapshot-metadata/pkg/api"
+	"github.com/kubernetes-csi/external-snapshot-metadata/pkg/iterator"
+	"github.com/sirupsen/logrus"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"k8s.io/client-go/rest"
+)
+
+func TestEmitterImplSnapshotMetadataIteratorRecord(t *testing.T) {
+	t.Run("records block metadata as ranges", func(t *testing.T) {
+		var got [][]Range
+		emitter := &emitterImpl{
+			logger: logrus.New(),
+			recordCallBack: func(ranges []Range) error {
+				got = append(got, ranges)
+				return nil
+			},
+		}
+
+		err := emitter.SnapshotMetadataIteratorRecord(7, iterator.IteratorMetadata{
+			BlockMetadata: []*api.BlockMetadata{
+				{ByteOffset: 10, SizeBytes: 20},
+				{ByteOffset: 40, SizeBytes: 50},
+			},
+		})
+
+		require.NoError(t, err)
+		assert.Equal(t, [][]Range{
+			{{Offset: 10, Length: 20}},
+			{{Offset: 40, Length: 50}},
+		}, got)
+	})
+
+	t.Run("rejects negative block metadata", func(t *testing.T) {
+		callbackCalled := false
+		emitter := &emitterImpl{
+			logger: logrus.New(),
+			recordCallBack: func(ranges []Range) error {
+				callbackCalled = true
+				return nil
+			},
+		}
+
+		err := emitter.SnapshotMetadataIteratorRecord(3, iterator.IteratorMetadata{
+			BlockMetadata: []*api.BlockMetadata{{ByteOffset: -1, SizeBytes: 20}},
+		})
+
+		require.Error(t, err)
+		assert.Contains(t, err.Error(), "invalid CBT metadata")
+		assert.False(t, callbackCalled)
+	})
+
+	t.Run("returns callback error", func(t *testing.T) {
+		wantErr := errors.New("record failed")
+		emitter := &emitterImpl{
+			logger: logrus.New(),
+			recordCallBack: func(ranges []Range) error {
+				return wantErr
+			},
+		}
+
+		err := emitter.SnapshotMetadataIteratorRecord(5, iterator.IteratorMetadata{
+			BlockMetadata: []*api.BlockMetadata{{ByteOffset: 1, SizeBytes: 2}},
+		})
+
+		require.ErrorIs(t, err, wantErr)
+	})
+}
+
+func TestEmitterImplSnapshotMetadataIteratorDone(t *testing.T) {
+	emitter := &emitterImpl{logger: logrus.New()}
+	assert.NoError(t, emitter.SnapshotMetadataIteratorDone(4))
+}
+
+func TestNewServiceInitializesFields(t *testing.T) {
+	logger := logrus.New()
+	cfg := &rest.Config{Host: "https://example.com"}
+
+	svc := NewService(logger, "velero-ns", "velero-sa", cfg)
+
+	impl, ok := svc.(*ServiceImpl)
+	require.True(t, ok)
+	assert.Same(t, logger, impl.logger)
+	assert.Equal(t, "velero-ns", impl.vsNamespace)
+	assert.Equal(t, "velero-sa", impl.SAName)
+	assert.Same(t, cfg, impl.clientConfig)
+}
+
+func TestNewServiceForwardsSANameToArgs(t *testing.T) {
+	originalBuildClients := buildClients
+	originalGetSnapshotMetadata := getSnapshotMetadata
+	t.Cleanup(func() {
+		buildClients = originalBuildClients
+		getSnapshotMetadata = originalGetSnapshotMetadata
+	})
+
+	buildClients = func(config *rest.Config) (iterator.Clients, error) {
+		return iterator.Clients{}, nil
+	}
+
+	var capturedArgs iterator.Args
+	getSnapshotMetadata = func(ctx context.Context, args iterator.Args) error {
+		capturedArgs = args
+		return nil
+	}
+
+	svc := NewService(logrus.New(), "velero-ns", "velero-sa", &rest.Config{Host: "https://example.com"})
+
+	err := svc.GetAllocatedBlocks(t.Context(), "snap-1", func([]Range) error { return nil })
+	require.NoError(t, err)
+	assert.Equal(t, "velero-ns", capturedArgs.Namespace)
+	assert.Equal(t, "velero-ns", capturedArgs.SANamespace)
+	assert.Equal(t, "velero-sa", capturedArgs.SAName)
+}
+
+func TestServiceImplGetAllocatedBlocks(t *testing.T) {
+	originalBuildClients := buildClients
+	originalGetSnapshotMetadata := getSnapshotMetadata
+	t.Cleanup(func() {
+		buildClients = originalBuildClients
+		getSnapshotMetadata = originalGetSnapshotMetadata
+	})
+
+	t.Run("build clients error is returned", func(t *testing.T) {
+		wantErr := errors.New("build clients failed")
+		getSnapshotCalled := false
+		buildClients = func(config *rest.Config) (iterator.Clients, error) {
+			return iterator.Clients{}, wantErr
+		}
+		getSnapshotMetadata = func(ctx context.Context, args iterator.Args) error {
+			getSnapshotCalled = true
+			return nil
+		}
+
+		service := &ServiceImpl{logger: logrus.New(), clientConfig: &rest.Config{Host: "https://example.com"}}
+
+		err := service.GetAllocatedBlocks(t.Context(), "snap-1", func([]Range) error { return nil })
+
+		require.ErrorIs(t, err, wantErr)
+		assert.False(t, getSnapshotCalled)
+	})
+
+	t.Run("forwards args and callback", func(t *testing.T) {
+		cfg := &rest.Config{Host: "https://example.com"}
+		fakeClients := iterator.Clients{}
+		var capturedArgs iterator.Args
+		var recorded [][]Range
+		buildClients = func(config *rest.Config) (iterator.Clients, error) {
+			assert.Same(t, cfg, config)
+			return fakeClients, nil
+		}
+		getSnapshotMetadata = func(ctx context.Context, args iterator.Args) error {
+			capturedArgs = args
+			return args.Emitter.SnapshotMetadataIteratorRecord(1, iterator.IteratorMetadata{
+				BlockMetadata: []*api.BlockMetadata{{ByteOffset: 11, SizeBytes: 22}},
+			})
+		}
+
+		service := &ServiceImpl{
+			logger:       logrus.New(),
+			vsNamespace:  "velero-ns",
+			SAName:       "sa-name",
+			clientConfig: cfg,
+		}
+
+		err := service.GetAllocatedBlocks(t.Context(), "snap-1", func(ranges []Range) error {
+			recorded = append(recorded, ranges)
+			return nil
+		})
+
+		require.NoError(t, err)
+		assert.Equal(t, fakeClients, capturedArgs.Clients)
+		assert.Equal(t, "snap-1", capturedArgs.SnapshotName)
+		assert.Empty(t, capturedArgs.PrevSnapshotName)
+		assert.Equal(t, "velero-ns", capturedArgs.Namespace)
+		assert.Equal(t, "velero-ns", capturedArgs.SANamespace)
+		assert.Equal(t, "sa-name", capturedArgs.SAName)
+		assert.Equal(t, iterator.DefaultTokenExpirySeconds, capturedArgs.TokenExpirySecs)
+		assert.Zero(t, capturedArgs.MaxResults)
+		assert.Equal(t, [][]Range{{{Offset: 11, Length: 22}}}, recorded)
+	})
+}
+
+func TestServiceImplGetChangedBlocks(t *testing.T) {
+	originalBuildClients := buildClients
+	originalGetSnapshotMetadata := getSnapshotMetadata
+	t.Cleanup(func() {
+		buildClients = originalBuildClients
+		getSnapshotMetadata = originalGetSnapshotMetadata
+	})
+
+	buildClients = func(config *rest.Config) (iterator.Clients, error) {
+		return iterator.Clients{}, nil
+	}
+
+	t.Run("sets previous snapshot name", func(t *testing.T) {
+		var capturedArgs iterator.Args
+		getSnapshotMetadata = func(ctx context.Context, args iterator.Args) error {
+			capturedArgs = args
+			return nil
+		}
+
+		service := &ServiceImpl{
+			logger:       logrus.New(),
+			vsNamespace:  "velero-ns",
+			SAName:       "sa-name",
+			clientConfig: &rest.Config{Host: "https://example.com"},
+		}
+
+		err := service.GetChangedBlocks(t.Context(), "snap-2", "snap-1", func([]Range) error { return nil })
+
+		require.NoError(t, err)
+		assert.Equal(t, "snap-2", capturedArgs.SnapshotName)
+		assert.Equal(t, "snap-1", capturedArgs.PrevSnapshotName)
+		assert.Equal(t, "velero-ns", capturedArgs.Namespace)
+		assert.Equal(t, iterator.DefaultTokenExpirySeconds, capturedArgs.TokenExpirySecs)
+		assert.Zero(t, capturedArgs.MaxResults)
+	})
+
+	t.Run("returns snapshot metadata error", func(t *testing.T) {
+		wantErr := errors.New("metadata failed")
+		getSnapshotMetadata = func(ctx context.Context, args iterator.Args) error {
+			return wantErr
+		}
+
+		service := &ServiceImpl{
+			logger:       logrus.New(),
+			clientConfig: &rest.Config{Host: "https://example.com"},
+		}
+
+		err := service.GetChangedBlocks(t.Context(), "snap-2", "snap-1", func([]Range) error { return nil })
+
+		require.ErrorIs(t, err, wantErr)
+	})
+}
--- a/pkg/cbtservice/service.go
+++ b/pkg/cbtservice/service.go
@@ -16,7 +16,9 @@ limitations under the License.

 package cbtservice

-import "context"
+import (
+	"context"
+)

 // Range defines the range of a change
 type Range struct {
--- a/pkg/cmd/cli/datamover/backup.go
+++ b/pkg/cmd/cli/datamover/backup.go
@@ -32,6 +32,7 @@ import (

 	"github.com/vmware-tanzu/velero/internal/credentials"
 	"github.com/vmware-tanzu/velero/pkg/buildinfo"
+	"github.com/vmware-tanzu/velero/pkg/cbtservice"
 	"github.com/vmware-tanzu/velero/pkg/client"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/signals"
 	"github.com/vmware-tanzu/velero/pkg/datamover"
@@ -56,6 +57,7 @@ type dataMoverBackupConfig struct {
 	volumeMode      string
 	duName          string
 	resourceTimeout time.Duration
+	cbtSAName       string
 }

 func NewBackupCommand(f client.Factory) *cobra.Command {
@@ -92,6 +94,7 @@ func NewBackupCommand(f client.Factory) *cobra.Command {
 	command.Flags().StringVar(&config.volumeMode, "volume-mode", config.volumeMode, "The mode of the volume to be backed up")
 	command.Flags().StringVar(&config.duName, "data-upload", config.duName, "The data upload name")
 	command.Flags().DurationVar(&config.resourceTimeout, "resource-timeout", config.resourceTimeout, "How long to wait for resource processes which are not covered by other specific timeout parameters.")
+	command.Flags().StringVar(&config.cbtSAName, "cbt-sa-name", config.cbtSAName, "The name of the service account used by CSI's CBT service")

 	_ = command.MarkFlagRequired("volume-path")
 	_ = command.MarkFlagRequired("volume-mode")
@@ -112,6 +115,7 @@ type dataMoverBackup struct {
 	config      dataMoverBackupConfig
 	kubeClient  kubernetes.Interface
 	dataPathMgr *datapath.Manager
+	cbtService  cbtservice.Service
 }

 func newdataMoverBackup(logger logrus.FieldLogger, factory client.Factory, config dataMoverBackupConfig) (*dataMoverBackup, error) {
@@ -197,6 +201,12 @@ func newdataMoverBackup(logger logrus.FieldLogger, factory client.Factory, confi
 		config:     config,
 		namespace:  factory.Namespace(),
 		nodeName:   nodeName,
+		cbtService: cbtservice.NewService(
+			logger,
+			factory.Namespace(),
+			config.cbtSAName,
+			clientConfig,
+		),
 	}

 	s.kubeClient, err = factory.KubeClient()
--- a/pkg/cmd/cli/datamover/backup_test.go
+++ b/pkg/cmd/cli/datamover/backup_test.go
@@ -22,14 +22,33 @@ import (
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
 	ctlclient "sigs.k8s.io/controller-runtime/pkg/client"

 	"github.com/vmware-tanzu/velero/internal/credentials"
+	"github.com/vmware-tanzu/velero/pkg/cbtservice"
+	factorymocks "github.com/vmware-tanzu/velero/pkg/client/mocks"
 	cacheMock "github.com/vmware-tanzu/velero/pkg/cmd/cli/datamover/mocks"
 	velerotest "github.com/vmware-tanzu/velero/pkg/test"
 	"github.com/vmware-tanzu/velero/pkg/util/filesystem"
 )

+func TestNewBackupCommandCBTSANameFlag(t *testing.T) {
+	f := factorymocks.NewFactory(t)
+	cmd := NewBackupCommand(f)
+
+	cbtSAFlag := cmd.Flags().Lookup("cbt-sa-name")
+	require.NotNil(t, cbtSAFlag)
+	assert.Empty(t, cbtSAFlag.DefValue)
+
+	err := cmd.Flags().Parse([]string{"--cbt-sa-name", "velero-cbt-sa"})
+	require.NoError(t, err)
+
+	flagValue, err := cmd.Flags().GetString("cbt-sa-name")
+	require.NoError(t, err)
+	assert.Equal(t, "velero-cbt-sa", flagValue)
+}
+
 func fakeCreateDataPathServiceWithErr(_ *dataMoverBackup) (dataPathService, error) {
 	return nil, errors.New("fake-create-data-path-error")
 }
@@ -129,6 +148,7 @@ func TestRunDataPath(t *testing.T) {
 				config: dataMoverBackupConfig{
 					duName: test.duName,
 				},
+				cbtService: cbtservice.Service(nil),
 			}

 			s.runDataPath()