cubefs

1
// Copyright 2022 The CubeFS Authors.
2
//
3
// Licensed under the Apache License, Version 2.0 (the "License");
4
// you may not use this file except in compliance with the License.
5
// You may obtain a copy of the License at
6
//
7
//     http://www.apache.org/licenses/LICENSE-2.0
8
//
9
// Unless required by applicable law or agreed to in writing, software
10
// distributed under the License is distributed on an "AS IS" BASIS,
11
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
12
// implied. See the License for the specific language governing
13
// permissions and limitations under the License.
14

15
package clustermgr
16

17
import (
18
	"context"
19
	"errors"
20
	"fmt"
21

22
	"github.com/cubefs/cubefs/blobstore/api/blobnode"
23
	"github.com/cubefs/cubefs/blobstore/common/proto"
24
	"github.com/cubefs/cubefs/blobstore/common/rpc"
25
)
26

27
type DiskInfoArgs struct {
28
	DiskID proto.DiskID `json:"disk_id"`
29
}
30

31
type DiskIDAllocRet struct {
32
	DiskID proto.DiskID `json:"disk_id"`
33
}
34

35
type DiskSetArgs struct {
36
	DiskID proto.DiskID     `json:"disk_id"`
37
	Status proto.DiskStatus `json:"status"`
38
}
39

40
type ListOptionArgs struct {
41
	Idc    string           `json:"idc,omitempty"`
42
	Rack   string           `json:"rack,omitempty"`
43
	Host   string           `json:"host,omitempty"`
44
	Status proto.DiskStatus `json:"status,omitempty"`
45
	// list disk info after marker
46
	Marker proto.DiskID `json:"marker,omitempty"`
47
	// one page count
48
	Count int `json:"count,omitempty"`
49
}
50

51
type ListDiskRet struct {
52
	Disks  []*blobnode.DiskInfo `json:"disks"`
53
	Marker proto.DiskID         `json:"marker"`
54
}
55

56
type DisksHeartbeatArgs struct {
57
	Disks []*blobnode.DiskHeartBeatInfo `json:"disks"`
58
}
59

60
type DisksHeartbeatRet struct {
61
	Disks []*DiskHeartbeatRet `json:"disks"`
62
}
63

64
type DiskHeartbeatRet struct {
65
	DiskID   proto.DiskID     `json:"disk_id"`
66
	Status   proto.DiskStatus `json:"status"`
67
	ReadOnly bool             `json:"read_only"`
68
}
69

70
type DiskStatInfo struct {
71
	IDC            string `json:"idc"`
72
	Total          int    `json:"total"`
73
	TotalChunk     int64  `json:"total_chunk"`
74
	TotalFreeChunk int64  `json:"total_free_chunk"`
75
	Available      int    `json:"available"`
76
	Readonly       int    `json:"readonly"`
77
	Expired        int    `json:"expired"`
78
	Broken         int    `json:"broken"`
79
	Repairing      int    `json:"repairing"`
80
	Repaired       int    `json:"repaired"`
81
	Dropping       int    `json:"dropping"`
82
	Dropped        int    `json:"dropped"`
83
}
84

85
type SpaceStatInfo struct {
86
	TotalSpace     int64          `json:"total_space"`
87
	FreeSpace      int64          `json:"free_space"`
88
	UsedSpace      int64          `json:"used_space"`
89
	WritableSpace  int64          `json:"writable_space"`
90
	TotalBlobNode  int64          `json:"total_blob_node"`
91
	TotalDisk      int64          `json:"total_disk"`
92
	DisksStatInfos []DiskStatInfo `json:"disk_stat_infos"`
93
}
94

95
type DiskAccessArgs struct {
96
	DiskID   proto.DiskID `json:"disk_id"`
97
	Readonly bool         `json:"readonly"`
98
}
99

100
// DiskIDAlloc alloc diskID from cluster manager
101
func (c *Client) AllocDiskID(ctx context.Context) (proto.DiskID, error) {
102
	ret := &DiskIDAllocRet{}
103
	err := c.PostWith(ctx, "/diskid/alloc", ret, rpc.NoneBody)
104
	if err != nil {
105
		return 0, err
106
	}
107
	return ret.DiskID, nil
108
}
109

110
// DiskInfo get disk info from cluster manager
111
func (c *Client) DiskInfo(ctx context.Context, id proto.DiskID) (ret *blobnode.DiskInfo, err error) {
112
	ret = &blobnode.DiskInfo{}
113
	err = c.GetWith(ctx, "/disk/info?disk_id="+id.ToString(), ret)
114
	return
115
}
116

117
// AddDisk add/register a new disk into cluster manager
118
func (c *Client) AddDisk(ctx context.Context, info *blobnode.DiskInfo) (err error) {
119
	err = c.PostWith(ctx, "/disk/add", nil, info)
120
	return
121
}
122

123
// SetDisk set disk status
124
func (c *Client) SetDisk(ctx context.Context, id proto.DiskID, status proto.DiskStatus) (err error) {
125
	if !status.IsValid() {
126
		return errors.New("invalid status")
127
	}
128
	return c.PostWith(ctx, "/disk/set", nil, &DiskSetArgs{DiskID: id, Status: status})
129
}
130

131
// ListHostDisk list specified host disk info from cluster manager
132
func (c *Client) ListHostDisk(ctx context.Context, host string) (ret []*blobnode.DiskInfo, err error) {
133
	listRet := ListDiskRet{}
134
	opt := &ListOptionArgs{Host: host, Count: 200}
135
	for {
136
		listRet, err = c.ListDisk(ctx, opt)
137
		if err != nil || len(listRet.Disks) == 0 {
138
			return
139
		}
140
		opt.Marker = listRet.Marker
141
		ret = append(ret, listRet.Disks...)
142
	}
143
}
144

145
// ListDisk list disk info from cluster manager
146
// when ListOptionArgs is default value, defalut return 10 diskInfos
147
func (c *Client) ListDisk(ctx context.Context, options *ListOptionArgs) (ret ListDiskRet, err error) {
148
	err = c.GetWith(ctx, fmt.Sprintf(
149
		"/disk/list?idc=%s&rack=%s&host=%s&status=%d&marker=%d&count=%d",
150
		options.Idc,
151
		options.Rack,
152
		options.Host,
153
		options.Status,
154
		options.Marker,
155
		options.Count,
156
	), &ret)
157
	return
158
}
159

160
// HeartbeatDisk report blobnode disk latest capacity info to cluster manager
161
func (c *Client) HeartbeatDisk(ctx context.Context, infos []*blobnode.DiskHeartBeatInfo) (ret []*DiskHeartbeatRet, err error) {
162
	result := &DisksHeartbeatRet{}
163
	args := &DisksHeartbeatArgs{Disks: infos}
164
	err = c.PostWith(ctx, "/disk/heartbeat", result, args)
165
	ret = result.Disks
166
	return
167
}
168

169
func (c *Client) DropDisk(ctx context.Context, id proto.DiskID) (err error) {
170
	err = c.PostWith(ctx, "/disk/drop", nil, &DiskInfoArgs{DiskID: id})
171
	return
172
}
173

174
func (c *Client) DroppedDisk(ctx context.Context, id proto.DiskID) (err error) {
175
	err = c.PostWith(ctx, "/disk/dropped", nil, &DiskInfoArgs{DiskID: id})
176
	return
177
}
178

179
func (c *Client) ListDroppingDisk(ctx context.Context) (ret []*blobnode.DiskInfo, err error) {
180
	result := &ListDiskRet{}
181
	err = c.GetWith(ctx, "/disk/droppinglist", result)
182
	ret = result.Disks
183
	return
184
}
185

186
func (c *Client) SetReadonlyDisk(ctx context.Context, id proto.DiskID, readonly bool) (err error) {
187
	err = c.PostWith(ctx, "/disk/access", nil, &DiskAccessArgs{DiskID: id, Readonly: readonly})
188
	return
189
}
190