Skip to content

Commit c7a6f6d

Browse files
BarkovBGBarkovBG
committed
[Disk Manager] Add ssd-nbs2 disks created via YDB NbsService
NBS 2.0 partitions are created with the same CreatePartition/DeletePartition gRPC API that dstool uses, instead of blockstore CreateVolume. First slice supports empty create and delete only.
1 parent fdae620 commit c7a6f6d

35 files changed

Lines changed: 1111 additions & 12 deletions

‎cloud/disk_manager/api/disk.proto‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,8 @@ enum DiskKind {
1515
DISK_KIND_SSD_MIRROR3 = 6;
1616
DISK_KIND_HDD_NONREPLICATED = 7;
1717
DISK_KIND_HDD_LOCAL = 8;
18+
// NBS 2.0 partition created via Ydb.Nbs.V1.NbsService.
19+
DISK_KIND_SSD_NBS2 = 9;
1820
}
1921

2022
enum DiskState {

‎cloud/disk_manager/internal/pkg/clients/ya.make‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,4 +2,5 @@ RECURSE(
22
metrics
33
nbs
44
nfs
5+
ydbnbs
56
)

‎cloud/disk_manager/internal/pkg/clients/ydbnbs/client.go‎

Lines changed: 266 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,266 @@
1+
package ydbnbs
2+
3+
import (
4+
"context"
5+
"fmt"
6+
"strings"
7+
"time"
8+
9+
"github.com/golang/protobuf/proto"
10+
ydbnbs_config "github.com/ydb-platform/nbs/cloud/disk_manager/internal/pkg/clients/ydbnbs/config"
11+
nbsprotos "github.com/ydb-platform/nbs/cloud/disk_manager/internal/pkg/clients/ydbnbs/protos"
12+
"github.com/ydb-platform/nbs/cloud/tasks/errors"
13+
"golang.org/x/exp/maps"
14+
"google.golang.org/grpc"
15+
"google.golang.org/grpc/credentials"
16+
"google.golang.org/grpc/credentials/insecure"
17+
)
18+
19+
////////////////////////////////////////////////////////////////////////////////
20+
21+
// Ydb.StatusIds.StatusCode.SUCCESS
22+
const ydbStatusSuccess int32 = 400000
23+
24+
type client struct {
25+
zoneID string
26+
timeout time.Duration
27+
dial func(ctx context.Context) (*grpc.ClientConn, error)
28+
}
29+
30+
func (c *client) ZoneID() string {
31+
return c.zoneID
32+
}
33+
34+
func (c *client) CreatePartition(
35+
ctx context.Context,
36+
params CreatePartitionParams,
37+
) (string, error) {
38+
39+
if len(params.DiskID) == 0 {
40+
return "", errors.NewNonRetriableErrorf("disk id is required")
41+
}
42+
if len(params.StoragePoolName) == 0 {
43+
return "", errors.NewNonRetriableErrorf(
44+
"storage pool name is required for ssd-nbs2 disk %v",
45+
params.DiskID,
46+
)
47+
}
48+
49+
req := &nbsprotos.CreatePartitionRequest{
50+
OperationParams: &nbsprotos.OperationParams{
51+
OperationMode: nbsprotos.OperationParams_SYNC,
52+
},
53+
DiskId: params.DiskID,
54+
BlockSize: params.BlockSize,
55+
BlocksCount: params.BlocksCount,
56+
StoragePoolName: params.StoragePoolName,
57+
StorageMedia: nbsprotos.StorageMediaKind_STORAGE_MEDIA_SSD,
58+
}
59+
60+
resp := &nbsprotos.CreatePartitionResponse{}
61+
err := c.invoke(ctx, "CreatePartition", req, resp)
62+
if err != nil {
63+
return "", err
64+
}
65+
66+
op, err := checkOperation(resp.GetOperation())
67+
if err != nil {
68+
return "", err
69+
}
70+
71+
result := &nbsprotos.CreatePartitionResult{}
72+
err = unpackOperationResult(op, result)
73+
if err != nil {
74+
return "", err
75+
}
76+
if len(result.GetTabletId()) == 0 {
77+
return "", errors.NewNonRetriableErrorf(
78+
"CreatePartition for disk %v returned empty tablet id",
79+
params.DiskID,
80+
)
81+
}
82+
83+
return result.GetTabletId(), nil
84+
}
85+
86+
func (c *client) DeletePartition(ctx context.Context, tabletID string) error {
87+
if len(tabletID) == 0 {
88+
return errors.NewNonRetriableErrorf("tablet id is required")
89+
}
90+
91+
req := &nbsprotos.DeletePartitionRequest{
92+
OperationParams: &nbsprotos.OperationParams{
93+
OperationMode: nbsprotos.OperationParams_SYNC,
94+
},
95+
TabletId: tabletID,
96+
}
97+
98+
resp := &nbsprotos.DeletePartitionResponse{}
99+
err := c.invoke(ctx, "DeletePartition", req, resp)
100+
if err != nil {
101+
return err
102+
}
103+
104+
_, err = checkOperation(resp.GetOperation())
105+
return err
106+
}
107+
108+
func (c *client) invoke(
109+
ctx context.Context,
110+
method string,
111+
req proto.Message,
112+
resp proto.Message,
113+
) error {
114+
115+
timeout := c.timeout
116+
if timeout <= 0 {
117+
timeout = 20 * time.Second
118+
}
119+
120+
ctx, cancel := context.WithTimeout(ctx, timeout)
121+
defer cancel()
122+
123+
conn, err := c.dial(ctx)
124+
if err != nil {
125+
return errors.NewRetriableError(err)
126+
}
127+
defer conn.Close()
128+
129+
stub := nbsprotos.NewNbsServiceClient(conn)
130+
switch method {
131+
case "CreatePartition":
132+
out, err := stub.CreatePartition(ctx, req.(*nbsprotos.CreatePartitionRequest))
133+
if err != nil {
134+
return errors.NewRetriableError(err)
135+
}
136+
proto.Merge(resp, out)
137+
case "DeletePartition":
138+
out, err := stub.DeletePartition(ctx, req.(*nbsprotos.DeletePartitionRequest))
139+
if err != nil {
140+
return errors.NewRetriableError(err)
141+
}
142+
proto.Merge(resp, out)
143+
default:
144+
return errors.NewNonRetriableErrorf("unknown nbs method %v", method)
145+
}
146+
147+
return nil
148+
}
149+
150+
func checkOperation(op *nbsprotos.Operation) (*nbsprotos.Operation, error) {
151+
if op == nil {
152+
return nil, errors.NewRetriableErrorf("empty operation in nbs response")
153+
}
154+
if !op.GetReady() {
155+
return nil, errors.NewRetriableErrorf(
156+
"nbs operation %v is not ready",
157+
op.GetId(),
158+
)
159+
}
160+
if op.GetStatus() != ydbStatusSuccess {
161+
return nil, errors.NewRetriableErrorf(
162+
"nbs operation %v failed with status %v",
163+
op.GetId(),
164+
op.GetStatus(),
165+
)
166+
}
167+
return op, nil
168+
}
169+
170+
func unpackOperationResult(op *nbsprotos.Operation, msg proto.Message) error {
171+
if op.GetResult() == nil {
172+
return errors.NewNonRetriableErrorf("nbs operation %v has empty result", op.GetId())
173+
}
174+
175+
err := proto.Unmarshal(op.GetResult().GetValue(), msg)
176+
if err != nil {
177+
return errors.NewNonRetriableErrorf(
178+
"failed to unpack nbs operation %v result: %w",
179+
op.GetId(),
180+
err,
181+
)
182+
}
183+
return nil
184+
}
185+
186+
////////////////////////////////////////////////////////////////////////////////
187+
188+
type factory struct {
189+
config *ydbnbs_config.ClientConfig
190+
timeout time.Duration
191+
}
192+
193+
func (f *factory) GetClient(ctx context.Context, zoneID string) (Client, error) {
194+
if f.config == nil {
195+
return nil, errors.NewNonRetriableErrorf(
196+
"ydb nbs client is not configured, available zones: []",
197+
)
198+
}
199+
200+
zone, ok := f.config.GetZones()[zoneID]
201+
if !ok {
202+
return nil, errors.NewNonRetriableErrorf(
203+
"unknown ydb nbs zone %q, available zones: %q",
204+
zoneID,
205+
maps.Keys(f.config.GetZones()),
206+
)
207+
}
208+
if len(zone.GetEndpoints()) == 0 {
209+
return nil, errors.NewNonRetriableErrorf(
210+
"no ydb nbs endpoints for zone %q",
211+
zoneID,
212+
)
213+
}
214+
215+
endpoint := normalizeEndpoint(zone.GetEndpoints()[0])
216+
creds, err := f.transportCredentials()
217+
if err != nil {
218+
return nil, err
219+
}
220+
221+
return &client{
222+
zoneID: zoneID,
223+
timeout: f.timeout,
224+
dial: func(ctx context.Context) (*grpc.ClientConn, error) {
225+
return grpc.DialContext(ctx, endpoint, grpc.WithTransportCredentials(creds))
226+
},
227+
}, nil
228+
}
229+
230+
func (f *factory) transportCredentials() (credentials.TransportCredentials, error) {
231+
if f.config.GetInsecure() || len(f.config.GetRootCertsFile()) == 0 {
232+
return insecure.NewCredentials(), nil
233+
}
234+
235+
creds, err := credentials.NewClientTLSFromFile(f.config.GetRootCertsFile(), "")
236+
if err != nil {
237+
return nil, errors.NewNonRetriableErrorf(
238+
"failed to load ydb nbs root certs from %v: %w",
239+
f.config.GetRootCertsFile(),
240+
err,
241+
)
242+
}
243+
return creds, nil
244+
}
245+
246+
func normalizeEndpoint(endpoint string) string {
247+
endpoint = strings.TrimPrefix(endpoint, "grpc://")
248+
endpoint = strings.TrimPrefix(endpoint, "grpcs://")
249+
return endpoint
250+
}
251+
252+
func NewFactory(config *ydbnbs_config.ClientConfig) (Factory, error) {
253+
timeout := 20 * time.Second
254+
if config != nil && len(config.GetRequestTimeout()) > 0 {
255+
parsed, err := time.ParseDuration(config.GetRequestTimeout())
256+
if err != nil {
257+
return nil, fmt.Errorf("invalid ydb nbs request timeout: %w", err)
258+
}
259+
timeout = parsed
260+
}
261+
262+
return &factory{
263+
config: config,
264+
timeout: timeout,
265+
}, nil
266+
}

‎cloud/disk_manager/internal/pkg/clients/ydbnbs/client_test.go‎

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
package ydbnbs
2+
3+
import (
4+
"testing"
5+
6+
"github.com/golang/protobuf/proto"
7+
"github.com/stretchr/testify/require"
8+
nbsprotos "github.com/ydb-platform/nbs/cloud/disk_manager/internal/pkg/clients/ydbnbs/protos"
9+
"google.golang.org/protobuf/types/known/anypb"
10+
)
11+
12+
func TestUnpackCreatePartitionResult(t *testing.T) {
13+
result := &nbsprotos.CreatePartitionResult{TabletId: "tablet-1"}
14+
raw, err := proto.Marshal(result)
15+
require.NoError(t, err)
16+
17+
op := &nbsprotos.Operation{
18+
Id: "op",
19+
Ready: true,
20+
Status: ydbStatusSuccess,
21+
Result: &anypb.Any{Value: raw},
22+
}
23+
24+
unpacked := &nbsprotos.CreatePartitionResult{}
25+
err = unpackOperationResult(op, unpacked)
26+
require.NoError(t, err)
27+
require.Equal(t, "tablet-1", unpacked.GetTabletId())
28+
}
29+
30+
func TestCheckOperationNotReady(t *testing.T) {
31+
_, err := checkOperation(&nbsprotos.Operation{Id: "op", Ready: false})
32+
require.Error(t, err)
33+
}
34+
35+
func TestCheckOperationFailedStatus(t *testing.T) {
36+
_, err := checkOperation(&nbsprotos.Operation{
37+
Id: "op",
38+
Ready: true,
39+
Status: 400030, // INTERNAL_ERROR
40+
})
41+
require.Error(t, err)
42+
}
43+
44+
func TestNormalizeEndpoint(t *testing.T) {
45+
require.Equal(t, "localhost:2135", normalizeEndpoint("grpc://localhost:2135"))
46+
require.Equal(t, "localhost:2135", normalizeEndpoint("grpcs://localhost:2135"))
47+
require.Equal(t, "localhost:2135", normalizeEndpoint("localhost:2135"))
48+
}

‎cloud/disk_manager/internal/pkg/clients/ydbnbs/config/config.proto‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,18 @@
1+
syntax = "proto2";
2+
3+
package ydbnbs;
4+
5+
option go_package = "github.com/ydb-platform/nbs/cloud/disk_manager/internal/pkg/clients/ydbnbs/config";
6+
7+
////////////////////////////////////////////////////////////////////////////////
8+
9+
message Zone {
10+
repeated string Endpoints = 1;
11+
}
12+
13+
message ClientConfig {
14+
map<string, Zone> Zones = 1;
15+
optional string RootCertsFile = 2;
16+
optional bool Insecure = 3 [default = true];
17+
optional string RequestTimeout = 4 [default = "20s"];
18+
}

‎cloud/disk_manager/internal/pkg/clients/ydbnbs/config/ya.make‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
PROTO_LIBRARY()
2+
3+
ONLY_TAGS(GO_PROTO)
4+
5+
SRCS(
6+
config.proto
7+
)
8+
9+
END()

‎cloud/disk_manager/internal/pkg/clients/ydbnbs/interface.go‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
package ydbnbs
2+
3+
import (
4+
"context"
5+
)
6+
7+
////////////////////////////////////////////////////////////////////////////////
8+
9+
type CreatePartitionParams struct {
10+
DiskID string
11+
BlockSize uint32
12+
BlocksCount uint64
13+
StoragePoolName string
14+
}
15+
16+
type Client interface {
17+
CreatePartition(ctx context.Context, params CreatePartitionParams) (tabletID string, err error)
18+
DeletePartition(ctx context.Context, tabletID string) error
19+
ZoneID() string
20+
}
21+
22+
type Factory interface {
23+
GetClient(ctx context.Context, zoneID string) (Client, error)
24+
}

0 commit comments

Comments
 (0)