gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat(clustermgr): add shardnode space and route api

Browse Source 
. #22433442

Signed-off-by: tangdeyi <tangdeyi@oppo.com>
This commit is contained in:
tangdeyi 2024-07-26 14:10:46 +08:00 committed by slasher
parent 552cb2fd9f
commit 3b348fa52c
11 changed files with 5255 additions and 341 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

96
blobstore/api/clustermgr/catalog.go Normal file
View File

@ -0,0 +1,96 @@
package clustermgr
import (
"bytes"
"context"
"crypto/md5"
"encoding/base64"
"encoding/binary"
"time"
)
const (
hashBytesLength = 16
authVersionV1 uint8 = 1
)
func (c *Client) CreateSpace(ctx context.Context, args *CreateSpaceArgs) (err error) {
err = c.PostWith(ctx, "/space/create", nil, args)
return
}
func (c *Client) GetSpace(ctx context.Context, args *GetSpaceArgs) (ret *Space, err error) {
ret = &Space{}
err = c.GetWith(ctx, "/space/get?name="+args.Name, ret)
return
}
func (c *Client) AuthSpace(ctx context.Context, args *AuthSpaceArgs) (err error) {
err = c.PostWith(ctx, "/space/auth", nil, args)
return
}
type AuthInfo struct {
AccessKey string
SecretKey string
}
// EncodeAuthInfo SDK generates token based on ak/sk
func EncodeAuthInfo(auth *AuthInfo) (token string, err error) {
timeStamp := time.Now().Unix()
hashBytes := CalculateHash(auth, timeStamp)
w := bytes.NewBuffer([]byte{})
if err = binary.Write(w, binary.LittleEndian, authVersionV1); err != nil {
return
}
if err = binary.Write(w, binary.LittleEndian, &timeStamp); err != nil {
return
}
if err = binary.Write(w, binary.LittleEndian, &hashBytes); err != nil {
return
}
return base64.URLEncoding.EncodeToString(w.Bytes()), nil
}
// DecodeAuthInfo server parses token
func DecodeAuthInfo(token string) (timestamp int64, hashBytes []byte, err error) {
b, err := base64.URLEncoding.DecodeString(token)
if err != nil {
return
}
var authVersion uint8
hashBytes = make([]byte, hashBytesLength)
r := bytes.NewBuffer(b)
if err = binary.Read(r, binary.LittleEndian, &authVersion); err != nil {
return
}
if authVersion == authVersionV1 {
if err = binary.Read(r, binary.LittleEndian, &timestamp); err != nil {
return
}
if err = binary.Read(r, binary.LittleEndian, &hashBytes); err != nil {
return
}
}
return
}
// CalculateHash server caculates hash based on ak/sk and timeStamp
func CalculateHash(auth *AuthInfo, timeStamp int64) (hashBytes []byte) {
b := make([]byte, 8)
binary.LittleEndian.PutUint64(b, uint64(timeStamp))
hash := md5.New()
hash.Write(b)
hash.Write([]byte(auth.AccessKey))
hash.Write([]byte(auth.SecretKey))
hashBytes = hash.Sum(nil)
return hashBytes
}
func (c *Client) GetCatalogChanges(ctx context.Context, args *GetCatalogChangesArgs) (ret *GetCatalogChangesRet, err error) {
ret = &GetCatalogChangesRet{}
err = c.PostWith(ctx, "/catalogchanges/get", ret, args)
return
}

2743
blobstore/api/clustermgr/catalog.pb.go Normal file
View File

File diff suppressed because it is too large Load Diff

71
blobstore/api/clustermgr/catalog.proto Normal file
View File

@ -0,0 +1,71 @@
syntax = "proto3";
package cubefs.blobstore.api.clustermgr;
option go_package = "./;clustermgr";
option (gogoproto.sizer_all) = true;
option (gogoproto.marshaler_all) = true;
option (gogoproto.unmarshaler_all) = true;
import "gogoproto/gogo.proto";
import "google/protobuf/any.proto";
import "cubefs/blobstore/api/clustermgr/shard.proto";
message Space {
uint32 space_id = 1 [(gogoproto.customname) = "SpaceID", (gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.SpaceID"];
string name = 2;
uint32 status = 3[(gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.SpaceStatus"];
repeated FieldMeta field_metas = 4 [(gogoproto.nullable) = false];
string access_key = 5;
string secret_key = 6;
}
message FieldMeta {
uint32 id = 1 [(gogoproto.customname) = "ID", (gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.FieldID"];
string name = 2;
uint32 field_type = 3 [(gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.FieldType"];
uint32 index_option = 4 [(gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.IndexOption"];
}
message CreateSpaceArgs {
string name = 1;
repeated FieldMeta field_metas = 2 [(gogoproto.nullable) = false];
}
message GetSpaceArgs {
string name = 1;
}
message AuthSpaceArgs {
string name = 1;
string token = 2;
}
message CatalogChangeShardAdd {
uint32 shard_id = 1 [(gogoproto.customname) = "ShardID", (gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.ShardID"];
uint64 epoch = 2;
repeated ShardUnitInfo units = 3 [(gogoproto.nullable) = false];
}
message CatalogChangeShardUpdate {
uint32 shard_id = 1 [(gogoproto.customname) = "ShardID", (gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.ShardID"];
uint64 epoch = 2;
ShardUnitInfo unit = 3 [(gogoproto.nullable) = false];
}
message CatalogChangeItem {
uint64 route_version = 1 [(gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.RouteVersion"];
uint32 type = 2 [(gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.CatalogChangeItemType"];
google.protobuf.Any item =3;
}
message GetCatalogChangesArgs {
uint64 route_version = 1 [(gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.RouteVersion"];
uint32 node_id = 2 [(gogoproto.customname) = "NodeID", (gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.NodeID"];
}
message GetCatalogChangesRet {
uint64 route_version = 1 [(gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.RouteVersion"];
repeated CatalogChangeItem items = 2 [(gogoproto.nullable) = false];
}

49
blobstore/api/clustermgr/disk.go
View File

@ -108,6 +108,11 @@ type ListDiskRet struct {
Marker proto.DiskID `json:"marker"`
}
type ListShardNodeDiskRet struct {
Disks []*ShardNodeDiskInfo `json:"disks"`
Marker proto.DiskID `json:"marker"`
}
type DisksHeartbeatArgs struct {
Disks []*DiskHeartBeatInfo `json:"disks"`
}
@ -251,13 +256,53 @@ func (c *Client) SetReadonlyDisk(ctx context.Context, id proto.DiskID, readonly
// AddShardNodeDisk add/register a new disk of shardnode into cluster manager
func (c *Client) AddShardNodeDisk(ctx context.Context, info *ShardNodeDiskInfo) (err error) {
err = c.PostWith(ctx, "shardnode/disk/add", nil, info)
err = c.PostWith(ctx, "/shardnode/disk/add", nil, info)
return
}
// HeartbeatShardNodeDisk report shardnode disk latest capacity info to cluster manager
func (c *Client) HeartbeatShardNodeDisk(ctx context.Context, infos []*ShardNodeDiskHeartbeatInfo) (err error) {
args := &ShardNodeDisksHeartbeatArgs{Disks: infos}
err = c.PostWith(ctx, "shardnode/disk/heartbeat", nil, args)
err = c.PostWith(ctx, "/shardnode/disk/heartbeat", nil, args)
return
}
// AllocShardNodeDiskID alloc shardnode diskID from cluster manager
func (c *Client) AllocShardNodeDiskID(ctx context.Context) (proto.DiskID, error) {
ret := &DiskIDAllocRet{}
err := c.PostWith(ctx, "/shardnode/diskid/alloc", ret, rpc.NoneBody)
if err != nil {
return 0, err
}
return ret.DiskID, nil
}
// ListShardNodeDisk list disk info from cluster manager
// when ListOptionArgs is default value, defalut return 10 diskInfos
func (c *Client) ListShardNodeDisk(ctx context.Context, options *ListOptionArgs) (ret ListShardNodeDiskRet, err error) {
err = c.GetWith(ctx, fmt.Sprintf(
"/shardnode/disk/list?idc=%s&rack=%s&host=%s&status=%d&marker=%d&count=%d",
options.Idc,
options.Rack,
options.Host,
options.Status,
options.Marker,
options.Count,
), &ret)
return
}
// ShardNodeDiskInfo get shardnode disk info from cluster manager
func (c *Client) ShardNodeDiskInfo(ctx context.Context, id proto.DiskID) (ret *ShardNodeDiskInfo, err error) {
ret = &ShardNodeDiskInfo{}
err = c.GetWith(ctx, "/shardnode/disk/info?disk_id="+id.ToString(), ret)
return
}
// SetShardNodeDisk set shardnode disk status
func (c *Client) SetShardNodeDisk(ctx context.Context, id proto.DiskID, status proto.DiskStatus) (err error) {
if !status.IsValid() {
return errors.New("invalid status")
}
return c.PostWith(ctx, "/shardnode/disk/set", nil, &DiskSetArgs{DiskID: id, Status: status})
}

14
blobstore/api/clustermgr/node.go
View File

@ -105,3 +105,17 @@ func (c *Client) AddShardNode(ctx context.Context, info *ShardNodeInfo) (proto.N
}
return ret.NodeID, nil
}
// ShardNodeInfo get shardnode info from cluster manager
func (c *Client) ShardNodeInfo(ctx context.Context, id proto.NodeID) (ret *ShardNodeInfo, err error) {
ret = &ShardNodeInfo{}
err = c.GetWith(ctx, "/shardnode/info?node_id="+id.ToString(), ret)
return
}
// ShardNodeTopoInfo get shardnode nodeset and diskset topo info from cluster manager
func (c *Client) ShardNodeTopoInfo(ctx context.Context) (ret *TopoInfo, err error) {
ret = &TopoInfo{}
err = c.GetWith(ctx, "/shardnode/topo/info", ret)
return
}

61
blobstore/api/clustermgr/shard.go
View File

@ -18,88 +18,47 @@ import (
"context"
"fmt"
"github.com/cubefs/cubefs/blobstore/common/proto"
"github.com/cubefs/cubefs/blobstore/common/rpc"
)
type AllocShardUnitArgs struct {
Suid proto.Suid `json:"suid"`
}
type AllocShardUnitRet struct {
Suid proto.Suid `json:"suid"`
DiskID proto.DiskID `json:"disk_id"`
Host string `json:"host"`
}
func (c *Client) AllocShardUnit(ctx context.Context, args *AllocShardUnitArgs) (ret *AllocShardUnitRet, err error) {
err = c.PostWith(ctx, "/shard/unit/alloc", &ret, args)
return
}
type UpdateShardArgs struct {
NewSuid proto.Suid `json:"new_suid"`
NewIsLeaner bool `json:"new_is_leaner"`
NewDiskID proto.DiskID `json:"new_disk_id"`
OldSuid proto.Suid `json:"old_suid"`
OldIsLeaner bool `json:"old_is_leaner"`
}
func (c *Client) UpdateShard(ctx context.Context, args *UpdateShardArgs) (err error) {
err = c.PostWith(ctx, "/shard/update", nil, args)
return
}
type ReportShardArgs struct {
ShardReports []ShardUnitInfo `json:"shard_reports"`
type ShardReportArgs struct {
ShardReport
}
type ReportShardRet struct {
ShardTasks []ShardTask `json:"shard_tasks"`
func (args *ShardReportArgs) Marshal() ([]byte, string, error) {
bytes, err := args.ShardReport.Marshal()
return bytes, rpc.MIMEStream, err
}
func (c *Client) ReportShard(ctx context.Context, args *ReportShardArgs) (ret []ShardTask, err error) {
result := &ReportShardRet{}
func (c *Client) ReportShard(ctx context.Context, args *ShardReportArgs) (ret []ShardTask, err error) {
result := &ShardReportRet{}
err = c.PostWith(ctx, "/shard/report", result, args)
return result.ShardTasks, err
}
type GetShardArgs struct {
ShardID proto.ShardID `json:"shard_id"`
}
func (c *Client) GetShardInfo(ctx context.Context, args *GetShardArgs) (ret *Shard, err error) {
ret = &Shard{}
err = c.GetWith(ctx, "/shard/get?shard_id="+args.ShardID.ToString(), ret)
return
}
type ListShardUnitArgs struct {
DiskID proto.DiskID `json:"disk_id"`
}
type ListShardUnitInfos struct {
ShardUnitInfos []*ShardUnitInfo `json:"shard_unit_infos"`
}
func (c *Client) ListShardUnit(ctx context.Context, args *ListShardUnitArgs) ([]*ShardUnitInfo, error) {
ret := &ListShardUnitInfos{}
ret := &ListShardUnitRet{}
err := c.GetWith(ctx, "/shard/unit/list?disk_id="+args.DiskID.ToString(), &ret)
return ret.ShardUnitInfos, err
}
type ListShardArgs struct {
// list marker
Marker proto.ShardID `json:"marker,omitempty"`
// one-page count
Count int `json:"count"`
}
type ListShards struct {
Shards []*Shard `json:"shards"`
Marker proto.ShardID `json:"marker"`
}
func (c *Client) ListShard(ctx context.Context, args *ListShardArgs) (ret ListShards, err error) {
func (c *Client) ListShard(ctx context.Context, args *ListShardArgs) (ret ListShardRet, err error) {
err = c.GetWith(ctx, fmt.Sprintf("/shard/list?marker=%d&count=%d", args.Marker, args.Count), &ret)
return
}

2443
blobstore/api/clustermgr/shard.pb.go
View File

File diff suppressed because it is too large Load Diff

54
blobstore/api/clustermgr/shard.proto
View File

@ -25,10 +25,6 @@ message ShardUnit {
bool learner = 3;
}
message ShardReportArgs {
repeated ShardUnitInfo shards = 1;
}
message ShardUnitInfo {
uint64 suid = 1 [(gogoproto.customname) = "Suid", (gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.Suid"];
uint32 disk_id = 2 [(gogoproto.customname) = "DiskID", (gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.DiskID"];
@ -36,6 +32,8 @@ message ShardUnitInfo {
uint32 leader_idx = 4;
cubefs.blobstore.common.sharding.Range range = 5 [(gogoproto.nullable) = false];
uint64 epoch = 6;
string host = 7;
bool learner = 8;
}
message ShardTask {
@ -44,3 +42,51 @@ message ShardTask {
uint32 suid = 3 [(gogoproto.customname) = "Suid", (gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.Suid"];
uint64 epoch = 4;
}
message ShardReport {
repeated ShardUnitInfo shards = 1;
}
message ShardReportRet {
repeated ShardTask shard_tasks = 1 [(gogoproto.nullable) = false];
}
message AllocShardUnitArgs{
uint64 suid = 1 [(gogoproto.customname) = "Suid", (gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.Suid"];
}
message AllocShardUnitRet {
uint64 suid = 1 [(gogoproto.customname) = "Suid", (gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.Suid"];
uint32 disk_id = 2 [(gogoproto.customname) = "DiskID", (gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.DiskID"];
string host = 3;
}
message UpdateShardArgs {
uint64 new_suid = 1 [(gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.Suid"];
uint32 new_disk_id = 2 [(gogoproto.customname) = "NewDiskID", (gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.DiskID"];
bool new_is_leaner = 3;
uint64 old_suid = 4 [(gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.Suid"];
bool old_is_leaner = 5;
}
message GetShardArgs {
uint32 shard_id = 1 [(gogoproto.customname) = "ShardID", (gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.ShardID"];
}
message ListShardUnitArgs {
uint32 disk_id = 1 [(gogoproto.customname) = "DiskID", (gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.DiskID"];
}
message ListShardUnitRet {
repeated ShardUnitInfo shard_unit_infos = 1;
}
message ListShardArgs {
uint32 marker = 1 [(gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.ShardID"];
uint32 count = 2;
}
message ListShardRet {
repeated Shard shards = 1;
uint32 Marker = 2 [(gogoproto.casttype) = "github.com/cubefs/cubefs/blobstore/common/proto.ShardID"];
}

55
blobstore/api/clustermgr/space.go
View File

@ -1,55 +0,0 @@
// Copyright 2024 The CubeFS Authors.
//
// 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 clustermgr
import (
"context"
"github.com/cubefs/cubefs/blobstore/common/proto"
)
type Space struct {
SpaceID proto.SpaceID `json:"sid"`
Name string `json:"name"`
Status proto.SpaceStatus `json:"status"`
FieldMetas []FieldMeta `json:"field_metas"`
}
type FieldMeta struct {
ID proto.FieldID `json:"id"`
Name string `json:"name"`
FieldType proto.FieldType `json:"field_type"`
IndexOption proto.IndexOption `json:"index_option"`
}
type CreateSpaceArgs struct {
Name string `json:"name"`
FieldMetas []FieldMeta `json:"field_metas"`
}
func (c *Client) CreateSpace(ctx context.Context, args *CreateSpaceArgs) (err error) {
err = c.PostWith(ctx, "/space/create", nil, args)
return
}
type GetSpaceArgs struct {
Name string `json:"name"`
}
func (c *Client) GetSpace(ctx context.Context, args *GetSpaceArgs) (ret *Space, err error) {
ret = &Space{}
err = c.GetWith(ctx, "/space/get?name="+args.Name, ret)
return
}

2
blobstore/common/proto/basic.go
View File

@ -31,6 +31,8 @@ type (
NodeSetID uint32
DiskSetID uint32
SpaceID uint64
RouteVersion uint64
)
func (id DiskID) Encode() []byte {

8
blobstore/common/proto/const.go
View File

@ -34,6 +34,8 @@ type (
NodeStatus uint8
DiskType uint8
NodeRole uint8
CatalogChangeItemType uint8
)
// disk status
@ -256,3 +258,9 @@ func (t TaskSwitch) String() string {
// MaxShardSize max size of a single shard
const MaxShardSize = 512 << 20
// catalog changeItem type
const (
CatalogChangeItemAddShard = CatalogChangeItemType(iota + 1)
CatalogChangeItemUpdateShard
)