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

init (fix conflicts and init proto packet)

Browse Source
This commit is contained in:
awzhgw 2019-03-19 19:28:02 +08:00
parent 69434112db
commit f9ddf3abf3
12 changed files with 1717 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
datanode/partition_metrics.go Normal file
View File

@ -0,0 +1,58 @@
// Copyright 2018 The Chubao 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 datanode
import "sync/atomic"
// DataPartitionMetrics defines the metrics related to the data partition
type DataPartitionMetrics struct {
WriteCnt uint64
ReadCnt uint64
TotalWriteLatency uint64
TotalReadLatency uint64
WriteLatency float64
ReadLatency float64
lastWriteLatency float64
lastReadLatency float64
}
// NewDataPartitionMetrics creates a new DataPartitionMetrics
func NewDataPartitionMetrics() *DataPartitionMetrics {
metrics := new(DataPartitionMetrics)
metrics.WriteCnt = 1
metrics.ReadCnt = 1
return metrics
}
// UpdateReadMetrics updates the read-related metrics
func (metrics *DataPartitionMetrics) UpdateReadMetrics(latency uint64) {
atomic.AddUint64(&metrics.ReadCnt, 1)
atomic.AddUint64(&metrics.TotalReadLatency, latency)
}
// UpdateWriteMetrics updates the write-related metrics
func (metrics *DataPartitionMetrics) UpdateWriteMetrics(latency uint64) {
atomic.AddUint64(&metrics.WriteCnt, 1)
atomic.AddUint64(&metrics.TotalWriteLatency, latency)
}
func (metrics *DataPartitionMetrics) recomputeLatency() {
metrics.ReadLatency = float64((atomic.LoadUint64(&metrics.TotalReadLatency)) / (atomic.LoadUint64(&metrics.ReadCnt)))
metrics.WriteLatency = float64((atomic.LoadUint64(&metrics.TotalWriteLatency)) / (atomic.LoadUint64(&metrics.WriteCnt)))
atomic.StoreUint64(&metrics.TotalReadLatency, 0)
atomic.StoreUint64(&metrics.TotalWriteLatency, 0)
atomic.StoreUint64(&metrics.WriteCnt, 1)
atomic.StoreUint64(&metrics.ReadCnt, 1)
}

9
datanode/server.go
View File

@ -184,7 +184,14 @@ func (s *DataNode) parseConfig(cfg *config.Config) (err error) {
}
s.port = port
if len(cfg.GetArray(ConfigKeyMasterAddr)) == 0 {
return ErrBadConfFile
return fmt.Errorf("Err:no port")
}
if !regexpPort.MatchString(port) {
return fmt.Errorf("Err:port must string")
}
s.port = port
if len(cfg.GetArray(ConfigKeyMasterAddr)) == 0 {
return fmt.Errorf("Err:masterAddr unavalid")
}
for _, ip := range cfg.GetArray(ConfigKeyMasterAddr) {
MasterHelper.AddNode(ip.(string))

21
datanode/wrap_operator.go
View File

@ -53,8 +53,8 @@ func (s *DataNode) OperatePacket(p *repl.Packet, c *net.TCPConn) (err error) {
logContent := fmt.Sprintf("action[OperatePacket] %v.",
p.LogMessage(p.GetOpMsg(), c.RemoteAddr().String(), start, nil))
switch p.Opcode {
case proto.OpStreamRead, proto.OpRead:
case proto.OpExtentRepairRead, proto.OpReadTinyDelete:
case proto.OpStreamRead, proto.OpRead, proto.OpExtentRepairRead:
case proto.OpReadTinyDelete:
log.LogRead(logContent)
case proto.OpWrite, proto.OpRandomWrite, proto.OpSyncRandomWrite, proto.OpSyncWrite:
log.LogWrite(logContent)
@ -414,6 +414,13 @@ func (s *DataNode) handleStreamReadPacket(p *repl.Packet, connect net.Conn, isRe
var (
err error
)
defer func() {
if err != nil {
logContent := fmt.Sprintf("action[OperatePacket] %v.",
p.LogMessage(p.GetOpMsg(), connect.RemoteAddr().String(), p.StartT, err))
log.LogError(logContent)
}
}()
partition := p.Object.(*DataPartition)
if !isRepairRead {
err = partition.CheckLeader(p, connect)
@ -441,7 +448,10 @@ func (s *DataNode) handleStreamReadPacket(p *repl.Packet, connect net.Conn, isRe
}
tpObject := exporter.NewTPCnt(p.GetOpMsg())
reply.ExtentOffset = offset
p.Size = uint32(currReadSize)
p.ExtentOffset = offset
reply.CRC, err = store.Read(reply.ExtentID, offset, int64(currReadSize), reply.Data, isRepairRead)
p.CRC = reply.CRC
tpObject.Set()
if err != nil {
reply.PackErrorBody(ActionStreamRead, err.Error())
@ -453,13 +463,14 @@ func (s *DataNode) handleStreamReadPacket(p *repl.Packet, connect net.Conn, isRe
}
reply.Size = uint32(currReadSize)
reply.ResultCode = proto.OpOk
p.ResultCode = proto.OpOk
logContent := fmt.Sprintf("action[OperatePacket] %v.",
p.LogMessage(p.GetOpMsg(), connect.RemoteAddr().String(), p.StartT, err))
log.LogRead(logContent)
if err = reply.WriteToConn(connect); err != nil {
p.PackErrorBody(ActionStreamRead, err.Error())
return
}
logContent := fmt.Sprintf("action[OperatePacket] %v.",
reply.LogMessage(reply.GetOpMsg(), connect.RemoteAddr().String(), reply.StartT, nil))
log.LogRead(logContent)
needReplySize -= currReadSize
offset += int64(currReadSize)
if currReadSize == util.ReadBlockSize {

367
proto/admin_proto.go Normal file
View File

@ -0,0 +1,367 @@
// Copyright 2018 The Chubao 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 proto
// api
const (
// Admin APIs
AdminGetCluster = "/admin/getCluster"
AdminGetDataPartition = "/dataPartition/get"
AdminLoadDataPartition = "/dataPartition/load"
AdminCreateDataPartition = "/dataPartition/create"
AdminDecommissionDataPartition = "/dataPartition/decommission"
AdminDeleteVol = "/vol/delete"
AdminUpdateVol = "/vol/update"
AdminCreateVol = "/admin/createVol"
AdminGetVol = "/admin/getVol"
AdminClusterFreeze = "/cluster/freeze"
AdminGetIP = "/admin/getIp"
AdminCreateMP = "/metaPartition/create"
AdminSetMetaNodeThreshold = "/threshold/set"
// Client APIs
ClientDataPartitions = "/client/partitions"
ClientVol = "/client/vol"
ClientMetaPartition = "/client/metaPartition"
ClientVolStat = "/client/volStat"
//raft node APIs
AddRaftNode = "/raftNode/add"
RemoveRaftNode = "/raftNode/remove"
// Node APIs
AddDataNode = "/dataNode/add"
DecommissionDataNode = "/dataNode/decommission"
DecommissionDisk = "/disk/decommission"
GetDataNode = "/dataNode/get"
AddMetaNode = "/metaNode/add"
DecommissionMetaNode = "/metaNode/decommission"
GetMetaNode = "/metaNode/get"
AdminLoadMetaPartition = "/metaPartition/load"
AdminDecommissionMetaPartition = "/metaPartition/decommission"
// Operation response
GetMetaNodeTaskResponse = "/metaNode/response" // Method: 'POST', ContentType: 'application/json'
GetDataNodeTaskResponse = "/dataNode/response" // Method: 'POST', ContentType: 'application/json'
GetTopologyView = "/topo/get"
)
// HTTPReply uniform response structure
type HTTPReply struct {
Code int32 `json:"code"`
Msg string `json:"msg"`
Data interface{} `json:"data"`
}
// RegisterMetaNodeResp defines the response to register a meta node.
type RegisterMetaNodeResp struct {
ID uint64
}
// ClusterInfo defines the cluster infomation.
type ClusterInfo struct {
Cluster string
Ip string
}
// CreateDataPartitionRequest defines the request to create a data partition.
type CreateDataPartitionRequest struct {
PartitionType string
PartitionId uint64
PartitionSize int
VolumeId string
IsRandomWrite bool
Members []Peer
}
// CreateDataPartitionResponse defines the response to the request of creating a data partition.
type CreateDataPartitionResponse struct {
PartitionId uint64
Status uint8
Result string
}
// DeleteDataPartitionRequest defines the request to delete a data partition.
type DeleteDataPartitionRequest struct {
DataPartitionType string
PartitionId uint64
PartitionSize int
}
// DeleteDataPartitionResponse defines the response to the request of deleting a data partition.
type DeleteDataPartitionResponse struct {
Status uint8
Result string
PartitionId uint64
}
// DataPartitionDecommissionRequest defines the request of decommissioning a data partition.
type DataPartitionDecommissionRequest struct {
PartitionId uint64
RemovePeer Peer
AddPeer Peer
}
// DataPartitionDecommissionResponse defines the response to the request of decommissioning a data partition.
type DataPartitionDecommissionResponse struct {
Status uint8
Result string
PartitionId uint64
}
// LoadDataPartitionRequest defines the request of loading a data partition.
type LoadDataPartitionRequest struct {
PartitionId uint64
}
// LoadDataPartitionResponse defines the response to the request of loading a data partition.
type LoadDataPartitionResponse struct {
PartitionId uint64
Used uint64
PartitionSnapshot []*File
Status uint8
PartitionStatus int
Result string
}
// File defines the file struct.
type File struct {
Name string
Crc uint32
Size uint32
Modified int64
}
// LoadMetaPartitionMetricRequest defines the request of loading the meta partition metrics.
type LoadMetaPartitionMetricRequest struct {
PartitionID uint64
Start uint64
End uint64
}
// LoadMetaPartitionMetricResponse defines the response to the request of loading the meta partition metrics.
type LoadMetaPartitionMetricResponse struct {
Start uint64
End uint64
MaxInode uint64
Status uint8
Result string
}
// HeartBeatRequest define the heartbeat request.
type HeartBeatRequest struct {
CurrTime int64
MasterAddr string
}
// PartitionReport defines the partition report.
type PartitionReport struct {
PartitionID uint64
PartitionStatus int
Total uint64
Used uint64
DiskPath string
IsLeader bool
ExtentCount int
NeedCompare bool
}
// DataNodeHeartbeatResponse defines the response to the data node heartbeat.
type DataNodeHeartbeatResponse struct {
Total uint64
Used uint64
Available uint64
TotalPartitionSize uint64 // volCnt * volsize
RemainingCapacity uint64 // remaining capacity to create partition
CreatedPartitionCnt uint32
MaxCapacity uint64 // maximum capacity to create partition
RackName string
PartitionReports []*PartitionReport
Status uint8
Result string
}
// MetaPartitionReport defines the meta partition report.
type MetaPartitionReport struct {
PartitionID uint64
Start uint64
End uint64
Status int
MaxInodeID uint64
IsLeader bool
}
// MetaNodeHeartbeatResponse defines the response to the meta node heartbeat request.
type MetaNodeHeartbeatResponse struct {
RackName string
Total uint64
Used uint64
MetaPartitionReports []*MetaPartitionReport
Status uint8
Result string
}
// DeleteFileRequest defines the request to delete a file.
type DeleteFileRequest struct {
VolId uint64
Name string
}
// DeleteFileResponse defines the response to the request of deleting a file.
type DeleteFileResponse struct {
Status uint8
Result string
VolId uint64
Name string
}
// DeleteMetaPartitionRequest defines the request of deleting a meta partition.
type DeleteMetaPartitionRequest struct {
PartitionID uint64
}
// DeleteMetaPartitionResponse defines the response to the request of deleting a meta partition.
type DeleteMetaPartitionResponse struct {
PartitionID uint64
Status uint8
Result string
}
// UpdateMetaPartitionRequest defines the request to update a meta partition.
type UpdateMetaPartitionRequest struct {
PartitionID uint64
VolName string
Start uint64
End uint64
}
// UpdateMetaPartitionResponse defines the response to the request of updating the meta partition.
type UpdateMetaPartitionResponse struct {
PartitionID uint64
VolName string
End uint64
Status uint8
Result string
}
// MetaPartitionDecommissionRequest defines the request of decommissioning a meta partition.
type MetaPartitionDecommissionRequest struct {
PartitionID uint64
VolName string
RemovePeer Peer
AddPeer Peer
}
// MetaPartitionDecommissionResponse defines the response to the request of decommissioning a meta partition.
type MetaPartitionDecommissionResponse struct {
PartitionID uint64
VolName string
Status uint8
Result string
}
// MetaPartitionLoadRequest defines the request to load meta partition.
type MetaPartitionLoadRequest struct {
PartitionID uint64
}
// MetaPartitionLoadResponse defines the response to the request of loading meta partition.
type MetaPartitionLoadResponse struct {
PartitionID uint64
DoCompare bool
ApplyID uint64
InodeSign uint32
DentrySign uint32
Addr string
}
// VolStatInfo defines the statistics related to a volume
type VolStatInfo struct {
Name string
TotalSize uint64
UsedSize uint64
}
// DataPartitionResponse defines the response from a data node to the master that is related to a data partition.
type DataPartitionResponse struct {
PartitionID uint64
Status int8
ReplicaNum uint8
Hosts []string
LeaderAddr string
}
// DataPartitionsView defines the view of a data partition
type DataPartitionsView struct {
DataPartitions []*DataPartitionResponse
}
func NewDataPartitionsView() (dataPartitionsView *DataPartitionsView) {
dataPartitionsView = new(DataPartitionsView)
dataPartitionsView.DataPartitions = make([]*DataPartitionResponse, 0)
return
}
// MetaPartitionView defines the view of a meta partition
type MetaPartitionView struct {
PartitionID uint64
Start uint64
End uint64
Members []string
LeaderAddr string
Status int8
}
// VolView defines the view of a volume
type VolView struct {
Name string
Status uint8
MetaPartitions []*MetaPartitionView
DataPartitions []*DataPartitionResponse
}
func NewVolView(name string, status uint8) (view *VolView) {
view = new(VolView)
view.Name = name
view.Status = status
view.MetaPartitions = make([]*MetaPartitionView, 0)
view.DataPartitions = make([]*DataPartitionResponse, 0)
return
}
func NewMetaPartitionView(partitionID, start, end uint64, status int8) (mpView *MetaPartitionView) {
mpView = new(MetaPartitionView)
mpView.PartitionID = partitionID
mpView.Start = start
mpView.End = end
mpView.Status = status
mpView.Members = make([]string, 0)
return
}
// SimpleVolView defines the simple view of a volume
type SimpleVolView struct {
ID uint64
Name string
Owner string
DpReplicaNum uint8
MpReplicaNum uint8
Status uint8
Capacity uint64 // GB
RwDpCnt int
MpCnt int
DpCnt int
}

124
proto/admin_task.go Normal file
View File

@ -0,0 +1,124 @@
// Copyright 2018 The Chubao 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 proto
import (
"fmt"
"time"
)
const (
TaskFailed = 2
TaskStart = 0
TaskSucceeds = 1
TaskRunning = 3
ResponseInterval = 5
ResponseTimeOut = 100
MaxSendCount = 5
)
// AdminTask defines the administration task.
type AdminTask struct {
ID string
OpCode uint8
OperatorAddr string
Status int8
SendTime int64
CreateTime int64
SendCount uint8
Request interface{}
Response interface{}
}
// ToString returns the string format of the task.
func (t *AdminTask) ToString() (msg string) {
msg = fmt.Sprintf("ID[%v] Status[%d] LastSendTime[%v] SendCount[%v] Request[%v] Response[%v]",
t.ID, t.Status, t.SendTime, t.SendCount, t.Request, t.Response)
return
}
// CheckTaskNeedSend checks if the task needs to be sent out.
func (t *AdminTask) CheckTaskNeedSend() (needRetry bool) {
if (int)(t.SendCount) < MaxSendCount && time.Now().Unix()-t.SendTime > (int64)(ResponseInterval) {
needRetry = true
}
return
}
// CheckTaskTimeOut checks if the task is timed out.
func (t *AdminTask) CheckTaskTimeOut() (notResponse bool) {
if (int)(t.SendCount) >= MaxSendCount || (t.SendTime > 0 && (time.Now().Unix()-t.SendTime > int64(ResponseTimeOut))) {
notResponse = true
}
return
}
// SetStatus sets the status of the task.
func (t *AdminTask) SetStatus(status int8) {
t.Status = status
}
// IsTaskSuccessful returns if the task has been executed successful.
func (t *AdminTask) IsTaskSuccessful() (isSuccess bool) {
if t.Status == TaskSucceeds {
isSuccess = true
}
return
}
// IsTaskFailed returns if the task failed.
func (t *AdminTask) IsTaskFailed() (isFail bool) {
if t.Status == TaskFailed {
isFail = true
}
return
}
// IsUrgentTask returns if the task is urgent.
func (t *AdminTask) IsUrgentTask() bool {
return t.isCreateTask() || t.isLoadTask() || t.isUpdateMetaPartitionTask()
}
// isUpdateMetaPartitionTask checks if the task is to update the meta partition.
func (t *AdminTask) isUpdateMetaPartitionTask() bool {
return t.OpCode == OpUpdateMetaPartition
}
func (t *AdminTask) isLoadTask() bool {
return t.OpCode == OpLoadDataPartition
}
func (t *AdminTask) isCreateTask() bool {
return t.OpCode == OpCreateDataPartition || t.OpCode == OpCreateMetaPartition
}
// IsHeartbeatTask returns if the task is a heartbeat task.
func (t *AdminTask) IsHeartbeatTask() bool {
return t.OpCode == OpDataNodeHeartbeat || t.OpCode == OpMetaNodeHeartbeat
}
// NewAdminTask returns a new adminTask.
func NewAdminTask(opCode uint8, opAddr string, request interface{}) (t *AdminTask) {
t = new(AdminTask)
t.OpCode = opCode
t.Request = request
t.OperatorAddr = opAddr
t.ID = fmt.Sprintf("addr[%v]_op[%v]", t.OperatorAddr, t.OpCode)
t.CreateTime = time.Now().Unix()
return
}

134
proto/errors.go Normal file
View File

@ -0,0 +1,134 @@
// Copyright 2018 The Chubao 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 proto
import (
"github.com/juju/errors"
)
//err
var (
ErrSuc = errors.New("success")
ErrInternalError = errors.New("internal error")
ErrParamError = errors.New("parameter error")
ErrInvalidCfg = errors.New("bad configuration file")
ErrPersistenceByRaft = errors.New("persistence by raft occurred error")
ErrMarshalData = errors.New("marshal data error")
ErrUnmarshalData = errors.New("unmarshal data error")
ErrVolNotExists = errors.New("vol not exists")
ErrMetaPartitionNotExists = errors.New("meta partition not exists")
ErrDataPartitionNotExists = errors.New("data partition not exists")
ErrDataNodeNotExists = errors.New("data partition not exists")
ErrMetaNodeNotExists = errors.New("data partition not exists")
ErrDuplicateVol = errors.New("duplicate vol")
ErrActiveDataNodesTooLess = errors.New("no enough active data node")
ErrActiveMetaNodesTooLess = errors.New("no enough active meta node")
ErrInvalidMpStart = errors.New("invalid meta partition start value")
ErrNoAvailDataPartition = errors.New("no available data partition")
ErrReshuffleArray = errors.New("the array to be reshuffled is nil")
ErrIllegalDataReplica = errors.New("data replica is illegal")
ErrMissingReplica = errors.New("a missing data replica is found")
ErrHasOneMissingReplica = errors.New("there is a missing data replica")
ErrNoDataNodeToWrite = errors.New("No data node available for creating a data partition")
ErrNoMetaNodeToWrite = errors.New("No meta node available for creating a meta partition")
ErrCannotBeOffLine = errors.New("cannot take the data replica offline")
ErrNoDataNodeToCreateDataPartition = errors.New("no enough data nodes for creating a data partition")
ErrNoRackToCreateDataPartition = errors.New("no rack available for creating a data partition")
ErrNoNodeSetToCreateDataPartition = errors.New("no node set available for creating a data partition")
ErrNoNodeSetToCreateMetaPartition = errors.New("no node set available for creating a meta partition")
ErrNoMetaNodeToCreateMetaPartition = errors.New("no enough meta nodes for creating a meta partition")
ErrIllegalMetaReplica = errors.New("illegal meta replica")
ErrNoEnoughReplica = errors.New("no enough replicas")
ErrNoLeader = errors.New("no leader")
ErrVolAuthKeyNotMatch = errors.New("client and server auth key do not match")
)
// http response error code and error message definitions
const (
ErrCodeSuccess = iota
ErrCodeInternalError
ErrCodeParamError
ErrCodeInvalidCfg
ErrCodePersistenceByRaft
ErrCodeMarshalData
ErrCodeUnmarshalData
ErrCodeVolNotExists
ErrCodeMetaPartitionNotExists
ErrCodeDataPartitionNotExists
ErrCodeDataNodeNotExists
ErrCodeMetaNodeNotExists
ErrCodeDuplicateVol
ErrCodeActiveDataNodesTooLess
ErrCodeActiveMetaNodesTooLess
ErrCodeInvalidMpStart
ErrCodeNoAvailDataPartition
ErrCodeReshuffleArray
ErrCodeIllegalDataReplica
ErrCodeMissingReplica
ErrCodeHasOneMissingReplica
ErrCodeNoDataNodeToWrite
ErrCodeNoMetaNodeToWrite
ErrCodeCannotBeOffLine
ErrCodeNoDataNodeToCreateDataPartition
ErrCodeNoRackToCreateDataPartition
ErrCodeNoNodeSetToCreateDataPartition
ErrCodeNoNodeSetToCreateMetaPartition
ErrCodeNoMetaNodeToCreateMetaPartition
ErrCodeIllegalMetaReplica
ErrCodeNoEnoughReplica
ErrCodeNoLeader
ErrCodeVolAuthKeyNotMatch
)
// Err2CodeMap error map to code
var Err2CodeMap = map[error]int32{
ErrSuc: ErrCodeSuccess,
ErrInternalError: ErrCodeInternalError,
ErrParamError: ErrCodeParamError,
ErrInvalidCfg: ErrCodeInvalidCfg,
ErrPersistenceByRaft: ErrCodePersistenceByRaft,
ErrMarshalData: ErrCodeMarshalData,
ErrUnmarshalData: ErrCodeUnmarshalData,
ErrVolNotExists: ErrCodeVolNotExists,
ErrMetaPartitionNotExists: ErrCodeMetaPartitionNotExists,
ErrDataPartitionNotExists: ErrCodeDataPartitionNotExists,
ErrDataNodeNotExists: ErrCodeDataNodeNotExists,
ErrMetaNodeNotExists: ErrCodeMetaNodeNotExists,
ErrDuplicateVol: ErrCodeDuplicateVol,
ErrActiveDataNodesTooLess: ErrCodeActiveDataNodesTooLess,
ErrActiveMetaNodesTooLess: ErrCodeActiveMetaNodesTooLess,
ErrInvalidMpStart: ErrCodeInvalidMpStart,
ErrNoAvailDataPartition: ErrCodeNoAvailDataPartition,
ErrReshuffleArray: ErrCodeReshuffleArray,
ErrIllegalDataReplica: ErrCodeIllegalDataReplica,
ErrMissingReplica: ErrCodeMissingReplica,
ErrHasOneMissingReplica: ErrCodeHasOneMissingReplica,
ErrNoDataNodeToWrite: ErrCodeNoDataNodeToWrite,
ErrNoMetaNodeToWrite: ErrCodeNoMetaNodeToWrite,
ErrCannotBeOffLine: ErrCodeCannotBeOffLine,
ErrNoDataNodeToCreateDataPartition: ErrCodeNoDataNodeToCreateDataPartition,
ErrNoRackToCreateDataPartition: ErrCodeNoRackToCreateDataPartition,
ErrNoNodeSetToCreateDataPartition: ErrCodeNoNodeSetToCreateDataPartition,
ErrNoNodeSetToCreateMetaPartition: ErrCodeNoNodeSetToCreateMetaPartition,
ErrNoMetaNodeToCreateMetaPartition: ErrCodeNoMetaNodeToCreateMetaPartition,
ErrIllegalMetaReplica: ErrCodeIllegalMetaReplica,
ErrNoEnoughReplica: ErrCodeNoEnoughReplica,
ErrNoLeader: ErrCodeNoLeader,
ErrVolAuthKeyNotMatch: ErrCodeVolAuthKeyNotMatch,
}

127
proto/extent_key.go Normal file
View File

@ -0,0 +1,127 @@
// Copyright 2018 The Chubao 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 proto
import (
"bytes"
"encoding/binary"
"errors"
"fmt"
"github.com/chubaofs/cfs/util/btree"
)
var (
ExtentLength = 40
InvalidKey = errors.New("invalid key error")
)
// ExtentKey defines the extent key struct.
type ExtentKey struct {
FileOffset uint64
PartitionId uint64
ExtentId uint64
ExtentOffset uint64
Size uint32
CRC uint32
}
// String returns the string format of the extentKey.
func (k ExtentKey) String() string {
return fmt.Sprintf("ExtentKey{FileOffset(%v),Partition(%v),ExtentID(%v),ExtentOffset(%v),Size(%v),CRC(%v)}", k.FileOffset, k.PartitionId, k.ExtentId, k.ExtentOffset, k.Size, k.CRC)
}
// Less defines the less comparator.
func (k *ExtentKey) Less(than btree.Item) bool {
that := than.(*ExtentKey)
return k.FileOffset < that.FileOffset
}
// Marshal marshals the extent key.
func (k *ExtentKey) Copy() btree.Item {
return k
}
func (k *ExtentKey) Marshal() (m string) {
return fmt.Sprintf("%v_%v_%v_%v_%v_%v", k.FileOffset, k.PartitionId, k.ExtentId, k.ExtentOffset, k.Size, k.CRC)
}
// MarshalBinary marshals the binary format of the extent key.
func (k *ExtentKey) MarshalBinary() ([]byte, error) {
buf := bytes.NewBuffer(make([]byte, 0, ExtentLength))
if err := binary.Write(buf, binary.BigEndian, k.FileOffset); err != nil {
return nil, err
}
if err := binary.Write(buf, binary.BigEndian, k.PartitionId); err != nil {
return nil, err
}
if err := binary.Write(buf, binary.BigEndian, k.ExtentId); err != nil {
return nil, err
}
if err := binary.Write(buf, binary.BigEndian, k.ExtentOffset); err != nil {
return nil, err
}
if err := binary.Write(buf, binary.BigEndian, k.Size); err != nil {
return nil, err
}
if err := binary.Write(buf, binary.BigEndian, k.CRC); err != nil {
return nil, err
}
return buf.Bytes(), nil
}
// UnmarshalBinary unmarshals the binary format of the extent key.
func (k *ExtentKey) UnmarshalBinary(buf *bytes.Buffer) (err error) {
if err = binary.Read(buf, binary.BigEndian, &k.FileOffset); err != nil {
return
}
if err = binary.Read(buf, binary.BigEndian, &k.PartitionId); err != nil {
return
}
if err = binary.Read(buf, binary.BigEndian, &k.ExtentId); err != nil {
return
}
if err = binary.Read(buf, binary.BigEndian, &k.ExtentOffset); err != nil {
return
}
if err = binary.Read(buf, binary.BigEndian, &k.Size); err != nil {
return
}
if err = binary.Read(buf, binary.BigEndian, &k.CRC); err != nil {
return
}
return
}
// TODO remove
func (k *ExtentKey) UnMarshal(m string) (err error) {
_, err = fmt.Sscanf(m, "%v_%v_%v_%v_%v_%v", &k.FileOffset, &k.PartitionId, &k.ExtentId, &k.ExtentOffset, &k.Size, &k.CRC)
return
}
// TODO remove
func (k *ExtentKey) GetExtentKey() (m string) {
return fmt.Sprintf("%v_%v_%v", k.PartitionId, k.ExtentId, k.ExtentOffset)
}
type TinyExtentDeleteRecord struct {
FileOffset uint64
PartitionId uint64
ExtentId uint64
ExtentOffset uint64
Size uint32
CRC uint32
TinyDeleteFileOffset int64
}

260
proto/fs_proto.go Normal file
View File

@ -0,0 +1,260 @@
// Copyright 2018 The Chubao 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 proto
import (
"fmt"
"os"
"time"
)
const (
RootIno = uint64(1)
)
// Mode returns the fileMode.
func Mode(osMode os.FileMode) uint32 {
return uint32(osMode)
}
// OsMode returns os.FileMode.
func OsMode(mode uint32) os.FileMode {
return os.FileMode(mode)
}
// IsRegular checks if the mode is regular.
func IsRegular(mode uint32) bool {
return OsMode(mode).IsRegular()
}
// IsDir checks if the mode is dir.
func IsDir(mode uint32) bool {
return OsMode(mode).IsDir()
}
// IsSymlink checks if the mode is symlink.
func IsSymlink(mode uint32) bool {
return OsMode(mode)&os.ModeSymlink != 0
}
// InodeInfo defines the inode struct.
type InodeInfo struct {
Inode uint64 `json:"ino"`
Mode uint32 `json:"mode"`
Nlink uint32 `json:"nlink"`
Size uint64 `json:"sz"`
Uid uint32 `json:"uid"`
Gid uint32 `json:"gid"`
Generation uint64 `json:"gen"`
ModifyTime time.Time `json:"mt"`
CreateTime time.Time `json:"ct"`
AccessTime time.Time `json:"at"`
Target []byte `json:"tgt"`
}
// String returns the string format of the inode.
func (info *InodeInfo) String() string {
return fmt.Sprintf("Inode(%v) Mode(%v) OsMode(%v) Nlink(%v) Size(%v) Uid(%v) Gid(%v) Gen(%v)", info.Inode, info.Mode, OsMode(info.Mode), info.Nlink, info.Size, info.Uid, info.Gid, info.Generation)
}
// Dentry defines the dentry struct.
type Dentry struct {
Name string `json:"name"`
Inode uint64 `json:"ino"`
Type uint32 `json:"type"`
}
// String returns the string format of the dentry.
func (d Dentry) String() string {
return fmt.Sprintf("Dentry{Name(%v),Inode(%v),Type(%v)}", d.Name, d.Inode, d.Type)
}
// CreateInodeRequest defines the request to create an inode.
type CreateInodeRequest struct {
VolName string `json:"vol"`
PartitionID uint64 `json:"pid"`
Mode uint32 `json:"mode"`
Target []byte `json:"tgt"`
}
// CreateInodeResponse defines the response to the request of creating an inode.
type CreateInodeResponse struct {
Info *InodeInfo `json:"info"`
}
// LinkInodeRequest defines the request to link an inode.
type LinkInodeRequest struct {
VolName string `json:"vol"`
PartitionID uint64 `json:"pid"`
Inode uint64 `json:"ino"`
}
// LinkInodeResponse defines the response to the request of linking an inode.
type LinkInodeResponse struct {
Info *InodeInfo `json:"info"`
}
// UnlinkInodeRequest defines the request to unlink an inode.
type UnlinkInodeRequest struct {
VolName string `json:"vol"`
PartitionID uint64 `json:"pid"`
Inode uint64 `json:"ino"`
}
// UnlinkInodeResponse defines the response to the request of unlinking an inode.
type UnlinkInodeResponse struct {
Info *InodeInfo `json:"info"`
}
// EvictInodeRequest defines the request to evict an inode.
type EvictInodeRequest struct {
VolName string `json:"vol"`
PartitionID uint64 `json:"pid"`
Inode uint64 `json:"ino"`
}
// CreateDentryRequest defines the request to create a dentry.
type CreateDentryRequest struct {
VolName string `json:"vol"`
PartitionID uint64 `json:"pid"`
ParentID uint64 `json:"pino"`
Inode uint64 `json:"ino"`
Name string `json:"name"`
Mode uint32 `json:"mode"`
}
// UpdateDentryRequest defines the request to update a dentry.
type UpdateDentryRequest struct {
VolName string `json:"vol"`
PartitionID uint64 `json:"pid"`
ParentID uint64 `json:"pino"`
Name string `json:"name"`
Inode uint64 `json:"ino"` // new inode number
}
// UpdateDentryResponse defines the response to the request of updating a dentry.
type UpdateDentryResponse struct {
Inode uint64 `json:"ino"` // old inode number
}
// DeleteDentryRequest define the request tp delete a dentry.
type DeleteDentryRequest struct {
VolName string `json:"vol"`
PartitionID uint64 `json:"pid"`
ParentID uint64 `json:"pino"`
Name string `json:"name"`
}
// DeleteDentryResponse defines the response to the request of deleting a dentry.
type DeleteDentryResponse struct {
Inode uint64 `json:"ino"`
}
// LookupRequest defines the request for lookup.
type LookupRequest struct {
VolName string `json:"vol"`
PartitionID uint64 `json:"pid"`
ParentID uint64 `json:"pino"`
Name string `json:"name"`
}
// LookupResponse defines the response for the loopup request.
type LookupResponse struct {
Inode uint64 `json:"ino"`
Mode uint32 `json:"mode"`
}
// InodeGetRequest defines the request to get the inode.
type InodeGetRequest struct {
VolName string `json:"vol"`
PartitionID uint64 `json:"pid"`
Inode uint64 `json:"ino"`
}
// InodeGetResponse defines the response to the InodeGetRequest.
type InodeGetResponse struct {
Info *InodeInfo `json:"info"`
}
// BatchInodeGetRequest defines the request to get the inode in batch.
type BatchInodeGetRequest struct {
VolName string `json:"vol"`
PartitionID uint64 `json:"pid"`
Inodes []uint64 `json:"inos"`
}
// BatchInodeGetResponse defines the response to the request of getting the inode in batch.
type BatchInodeGetResponse struct {
Infos []*InodeInfo `json:"infos"`
}
// ReadDirRequest defines the request to read dir.
type ReadDirRequest struct {
VolName string `json:"vol"`
PartitionID uint64 `json:"pid"`
ParentID uint64 `json:"pino"`
}
// ReadDirResponse defines the response to the request of reading dir.
type ReadDirResponse struct {
Children []Dentry `json:"children"`
}
// AppendExtentKeyRequest defines the request to append an extent key.
type AppendExtentKeyRequest struct {
VolName string `json:"vol"`
PartitionID uint64 `json:"pid"`
Inode uint64 `json:"ino"`
Extent ExtentKey `json:"ek"`
}
// GetExtentsRequest defines the reques to get extents.
type GetExtentsRequest struct {
VolName string `json:"vol"`
PartitionID uint64 `json:"pid"`
Inode uint64 `json:"ino"`
}
// GetExtentsResponse defines the response to the request of getting extents.
type GetExtentsResponse struct {
Generation uint64 `json:"gen"`
Size uint64 `json:"sz"`
Extents []ExtentKey `json:"eks"`
}
// TruncateRequest defines the request to truncate.
type TruncateRequest struct {
VolName string `json:"vol"`
PartitionID uint64 `json:"pid"`
Inode uint64 `json:"ino"`
Size uint64 `json:"sz"`
}
// SetAttrRequest defines the request to set attribute.
type SetAttrRequest struct {
VolName string `json:"vol"`
PartitionID uint64 `json:"pid"`
Inode uint64 `json:"ino"`
Mode uint32 `json:"mode"`
Uid uint32 `json:"uid"`
Gid uint32 `json:"gid"`
Valid uint32 `json:"valid"`
}
const (
AttrMode uint32 = 1 << iota
AttrUid
AttrGid
)

50
proto/meta_proto.go Normal file
View File

@ -0,0 +1,50 @@
// Copyright 2018 The Chubao 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 proto
// CreateNameSpaceRequest defines the request to create a name space.
type CreateNameSpaceRequest struct {
Name string
}
// CreateNameSpaceResponse defines the response to the request of creating a name space.
type CreateNameSpaceResponse struct {
Status int
Result string
}
// Peer defines the peer of the node id and address.
type Peer struct {
ID uint64 `json:"id"`
Addr string `json:"addr"`
}
// CreateMetaPartitionRequest defines the request to create a meta partition.
type CreateMetaPartitionRequest struct {
MetaId string
VolName string
Start uint64
End uint64
PartitionID uint64
Members []Peer
}
// CreateMetaPartitionResponse defines the response to the request of creating a meta partition.
type CreateMetaPartitionResponse struct {
VolName string
PartitionID uint64
Status uint8
Result string
}

551
proto/packet.go Normal file
View File

@ -0,0 +1,551 @@
// Copyright 2018 The Chubao 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 proto
import (
"encoding/binary"
"encoding/json"
"errors"
"fmt"
"github.com/chubaofs/cfs/util"
"github.com/chubaofs/cfs/util/buf"
"io"
"net"
"strconv"
"sync/atomic"
"time"
)
var (
GRequestID = int64(1)
Buffers = buf.NewBufferPool()
)
// GenerateRequestID generates the request ID.
func GenerateRequestID() int64 {
return atomic.AddInt64(&GRequestID, 1)
}
const (
AddrSplit = "/"
)
// Operations
const (
ProtoMagic uint8 = 0xFF
OpInitResultCode uint8 = 0x00
OpCreateExtent uint8 = 0x01
OpMarkDelete uint8 = 0x02
OpWrite uint8 = 0x03
OpRead uint8 = 0x04
OpStreamRead uint8 = 0x05
OpGetAllWatermarks uint8 = 0x07
OpNotifyReplicasToRepair uint8 = 0x08
OpExtentRepairRead uint8 = 0x09
OpBroadcastMinAppliedID uint8 = 0x0A
OpRandomWrite uint8 = 0x0F
OpGetAppliedId uint8 = 0x10
OpGetPartitionSize uint8 = 0x11
OpSyncRandomWrite uint8 = 0x12
OpSyncWrite uint8 = 0x13
OpReadTinyDelete uint8 = 0x14
// Operations: Client -> MetaNode.
OpMetaCreateInode uint8 = 0x20
OpMetaUnlinkInode uint8 = 0x21
OpMetaCreateDentry uint8 = 0x22
OpMetaDeleteDentry uint8 = 0x23
OpMetaOpen uint8 = 0x24
OpMetaLookup uint8 = 0x25
OpMetaReadDir uint8 = 0x26
OpMetaInodeGet uint8 = 0x27
OpMetaBatchInodeGet uint8 = 0x28
OpMetaExtentsAdd uint8 = 0x29
OpMetaExtentsDel uint8 = 0x2A
OpMetaExtentsList uint8 = 0x2B
OpMetaUpdateDentry uint8 = 0x2C
OpMetaTruncate uint8 = 0x2D
OpMetaLinkInode uint8 = 0x2E
OpMetaEvictInode uint8 = 0x2F
OpMetaSetattr uint8 = 0x30
OpMetaReleaseOpen uint8 = 0x31
// Operations: Master -> MetaNode
OpCreateMetaPartition uint8 = 0x40
OpMetaNodeHeartbeat uint8 = 0x41
OpDeleteMetaPartition uint8 = 0x42
OpUpdateMetaPartition uint8 = 0x43
OpLoadMetaPartition uint8 = 0x44
OpDecommissionMetaPartition uint8 = 0x45
// Operations: Master -> DataNode
OpCreateDataPartition uint8 = 0x60
OpDeleteDataPartition uint8 = 0x61
OpLoadDataPartition uint8 = 0x62
OpDataNodeHeartbeat uint8 = 0x63
OpReplicateFile uint8 = 0x64
OpDeleteFile uint8 = 0x65
OpDecommissionDataPartition uint8 = 0x66
// Commons
OpIntraGroupNetErr uint8 = 0xF3
OpArgMismatchErr uint8 = 0xF4
OpNotExistErr uint8 = 0xF5
OpDiskNoSpaceErr uint8 = 0xF6
OpDiskErr uint8 = 0xF7
OpErr uint8 = 0xF8
OpAgain uint8 = 0xF9
OpExistErr uint8 = 0xFA
OpInodeFullErr uint8 = 0xFB
OpNotLeaderErr uint8 = 0xFC
OpNotPerm uint8 = 0xFD
OpNotEmtpy uint8 = 0xFE
OpOk uint8 = 0xF0
OpPing uint8 = 0xFF
)
const (
WriteDeadlineTime = 5
ReadDeadlineTime = 5
SyncSendTaskDeadlineTime = 20
NoReadDeadlineTime = -1
GetAllWatermarksDeadLineTime = 60
)
const (
TinyExtentType = 0
NormalExtentType = 1
)
// Packet defines the packet structure.
type Packet struct {
Magic uint8
ExtentType uint8
Opcode uint8
ResultCode uint8
RemainingFollowers uint8
CRC uint32
Size uint32
ArgLen uint32
KernelOffset uint64
PartitionID uint64
ExtentID uint64
ExtentOffset int64
ReqID int64
Arg []byte // for create or append ops, the data contains the address
Data []byte
StartT int64
}
// NewPacket returns a new packet.
func NewPacket() *Packet {
p := new(Packet)
p.Magic = ProtoMagic
p.StartT = time.Now().UnixNano()
return p
}
// NewPacketReqID returns a new packet with ReqID assigned.
func NewPacketReqID() *Packet {
p := NewPacket()
p.ReqID = GenerateRequestID()
return p
}
func (p *Packet) String() string {
return fmt.Sprintf("ReqID(%v)Op(%v)PartitionID(%v)ResultCode(%v)", p.ReqID, p.GetOpMsg(), p.PartitionID, p.GetResultMsg())
}
// GetStoreType returns the store type.
func (p *Packet) GetStoreType() (m string) {
switch p.ExtentType {
case TinyExtentType:
m = "TinyExtent"
case NormalExtentType:
m = "NormalExtent"
default:
m = "Unknown"
}
return
}
// GetOpMsg returns the operation type.
func (p *Packet) GetOpMsg() (m string) {
switch p.Opcode {
case OpCreateExtent:
m = "OpCreateExtent"
case OpMarkDelete:
m = "OpMarkDelete"
case OpWrite:
m = "OpWrite"
case OpRandomWrite:
m = "OpRandomWrite"
case OpRead:
m = "Read"
case OpStreamRead:
m = "OpStreamRead"
case OpGetAllWatermarks:
m = "OpGetAllWatermarks"
case OpNotifyReplicasToRepair:
m = "OpNotifyReplicasToRepair"
case OpExtentRepairRead:
m = "OpExtentRepairRead"
case OpIntraGroupNetErr:
m = "IntraGroupNetErr"
case OpMetaCreateInode:
m = "OpMetaCreateInode"
case OpMetaUnlinkInode:
m = "OpMetaUnlinkInode"
case OpMetaCreateDentry:
m = "OpMetaCreateDentry"
case OpMetaDeleteDentry:
m = "OpMetaDeleteDentry"
case OpMetaOpen:
m = "OpMetaOpen"
case OpMetaReleaseOpen:
m = "OpMetaReleaseOpen"
case OpMetaLookup:
m = "OpMetaLookup"
case OpMetaReadDir:
m = "OpMetaReadDir"
case OpMetaInodeGet:
m = "OpMetaInodeGet"
case OpMetaBatchInodeGet:
m = "OpMetaBatchInodeGet"
case OpMetaExtentsAdd:
m = "OpMetaExtentsAdd"
case OpMetaExtentsDel:
m = "OpMetaExtentsDel"
case OpMetaExtentsList:
m = "OpMetaExtentsList"
case OpMetaUpdateDentry:
m = "OpMetaUpdateDentry"
case OpMetaTruncate:
m = "OpMetaTruncate"
case OpMetaLinkInode:
m = "OpMetaLinkInode"
case OpMetaEvictInode:
m = "OpMetaEvictInode"
case OpMetaSetattr:
m = "OpMetaSetattr"
case OpCreateMetaPartition:
m = "OpCreateMetaPartition"
case OpMetaNodeHeartbeat:
m = "OpMetaNodeHeartbeat"
case OpDeleteMetaPartition:
m = "OpDeleteMetaPartition"
case OpUpdateMetaPartition:
m = "OpUpdateMetaPartition"
case OpLoadMetaPartition:
m = "OpLoadMetaPartition"
case OpDecommissionMetaPartition:
m = "OpDecommissionMetaPartition"
case OpCreateDataPartition:
m = "OpCreateDataPartition"
case OpDeleteDataPartition:
m = "OpDeleteDataPartition"
case OpLoadDataPartition:
m = "OpLoadDataPartition"
case OpDecommissionDataPartition:
m = "OpDecommissionDataPartition"
case OpDataNodeHeartbeat:
m = "OpDataNodeHeartbeat"
case OpReplicateFile:
m = "OpReplicateFile"
case OpDeleteFile:
m = "OpDeleteFile"
case OpGetAppliedId:
m = "OpGetAppliedId"
case OpGetPartitionSize:
m = "OpGetPartitionSize"
case OpSyncWrite:
m = "OpSyncWrite"
case OpSyncRandomWrite:
m = "OpSyncRandomWrite"
case OpReadTinyDelete:
m = "OpReadTinyDelete"
case OpPing:
m = "OpPing"
case OpBroadcastMinAppliedID:
m = "OpBroadcastMinAppliedID"
}
return
}
// GetResultMsg returns the result message.
func (p *Packet) GetResultMsg() (m string) {
if p == nil {
return ""
}
switch p.ResultCode {
case OpIntraGroupNetErr:
m = "IntraGroupNetErr"
case OpDiskNoSpaceErr:
m = "DiskNoSpaceErr"
case OpDiskErr:
m = "DiskErr"
case OpErr:
m = "Err"
case OpAgain:
m = "Again"
case OpOk:
m = "Ok"
case OpExistErr:
m = "ExistErr"
case OpInodeFullErr:
m = "InodeFullErr"
case OpArgMismatchErr:
m = "ArgUnmatchErr"
case OpNotExistErr:
m = "NotExistErr"
case OpNotLeaderErr:
m = "NotLeaderErr"
case OpNotPerm:
m = "NotPerm"
case OpNotEmtpy:
m = "DirNotEmpty"
default:
return fmt.Sprintf("Unknown ResultCode(%v)", p.ResultCode)
}
return
}
func (p *Packet) GetReqID() int64 {
return p.ReqID
}
// MarshalHeader marshals the packet header.
func (p *Packet) MarshalHeader(out []byte) {
out[0] = p.Magic
out[1] = p.ExtentType
out[2] = p.Opcode
out[3] = p.ResultCode
out[4] = p.RemainingFollowers
binary.BigEndian.PutUint32(out[5:9], p.CRC)
binary.BigEndian.PutUint32(out[9:13], p.Size)
binary.BigEndian.PutUint32(out[13:17], p.ArgLen)
binary.BigEndian.PutUint64(out[17:25], p.PartitionID)
binary.BigEndian.PutUint64(out[25:33], p.ExtentID)
binary.BigEndian.PutUint64(out[33:41], uint64(p.ExtentOffset))
binary.BigEndian.PutUint64(out[41:49], uint64(p.ReqID))
binary.BigEndian.PutUint64(out[49:util.PacketHeaderSize], p.KernelOffset)
return
}
// UnmarshalHeader unmarshals the packet header.
func (p *Packet) UnmarshalHeader(in []byte) error {
p.Magic = in[0]
if p.Magic != ProtoMagic {
return errors.New("Bad Magic " + strconv.Itoa(int(p.Magic)))
}
p.ExtentType = in[1]
p.Opcode = in[2]
p.ResultCode = in[3]
p.RemainingFollowers = in[4]
p.CRC = binary.BigEndian.Uint32(in[5:9])
p.Size = binary.BigEndian.Uint32(in[9:13])
p.ArgLen = binary.BigEndian.Uint32(in[13:17])
p.PartitionID = binary.BigEndian.Uint64(in[17:25])
p.ExtentID = binary.BigEndian.Uint64(in[25:33])
p.ExtentOffset = int64(binary.BigEndian.Uint64(in[33:41]))
p.ReqID = int64(binary.BigEndian.Uint64(in[41:49]))
p.KernelOffset = binary.BigEndian.Uint64(in[49:util.PacketHeaderSize])
return nil
}
// MarshalData marshals the packet data.
func (p *Packet) MarshalData(v interface{}) error {
data, err := json.Marshal(v)
if err == nil {
p.Data = data
p.Size = uint32(len(p.Data))
}
return err
}
// UnmarshalData unmarshals the packet data.
func (p *Packet) UnmarshalData(v interface{}) error {
return json.Unmarshal(p.Data, v)
}
// WriteToNoDeadLineConn writes through the connection without deadline.
func (p *Packet) WriteToNoDeadLineConn(c net.Conn) (err error) {
header, err := Buffers.Get(util.PacketHeaderSize)
if err != nil {
header = make([]byte, util.PacketHeaderSize)
}
defer Buffers.Put(header)
p.MarshalHeader(header)
if _, err = c.Write(header); err == nil {
if _, err = c.Write(p.Arg[:int(p.ArgLen)]); err == nil {
if p.Data != nil {
_, err = c.Write(p.Data[:p.Size])
}
}
}
return
}
// WriteToConn writes through the given connection.
func (p *Packet) WriteToConn(c net.Conn) (err error) {
c.SetWriteDeadline(time.Now().Add(WriteDeadlineTime * time.Second))
header, err := Buffers.Get(util.PacketHeaderSize)
if err != nil {
header = make([]byte, util.PacketHeaderSize)
}
defer Buffers.Put(header)
p.MarshalHeader(header)
if _, err = c.Write(header); err == nil {
if _, err = c.Write(p.Arg[:int(p.ArgLen)]); err == nil {
if p.Data != nil && p.Size != 0 {
_, err = c.Write(p.Data[:p.Size])
}
}
}
return
}
// ReadFull is a wrapper function of io.ReadFull.
func ReadFull(c net.Conn, buf *[]byte, readSize int) (err error) {
*buf = make([]byte, readSize)
_, err = io.ReadFull(c, (*buf)[:readSize])
return
}
// ReadFromConn reads the data from the given connection.
func (p *Packet) ReadFromConn(c net.Conn, timeoutSec int) (err error) {
if timeoutSec != NoReadDeadlineTime {
c.SetReadDeadline(time.Now().Add(time.Second * time.Duration(timeoutSec)))
} else {
c.SetReadDeadline(time.Time{})
}
header, err := Buffers.Get(util.PacketHeaderSize)
if err != nil {
header = make([]byte, util.PacketHeaderSize)
}
defer Buffers.Put(header)
if _, err = io.ReadFull(c, header); err != nil {
return
}
if err = p.UnmarshalHeader(header); err != nil {
return
}
if p.ArgLen > 0 {
p.Arg = make([]byte, int(p.ArgLen))
if _, err = io.ReadFull(c, p.Arg[:int(p.ArgLen)]); err != nil {
return err
}
}
if p.Size < 0 {
return
}
size := p.Size
if (p.Opcode == OpRead || p.Opcode == OpStreamRead || p.Opcode == OpExtentRepairRead) && p.ResultCode == OpInitResultCode {
size = 0
}
p.Data = make([]byte, size)
_, err = io.ReadFull(c, p.Data[:size])
return err
}
// PacketOkReply sets the result code as OpOk, and sets the body as empty.
func (p *Packet) PacketOkReply() {
p.ResultCode = OpOk
p.Size = 0
p.Data = nil
p.ArgLen = 0
}
// PacketOkWithBody sets the result code as OpOk, and sets the body with the give data.
func (p *Packet) PacketOkWithBody(reply []byte) {
p.Size = uint32(len(reply))
p.Data = make([]byte, p.Size)
copy(p.Data[:p.Size], reply)
p.ResultCode = OpOk
p.ArgLen = 0
}
// PacketErrorWithBody sets the packet with error code whose body is filled with the given data.
func (p *Packet) PacketErrorWithBody(code uint8, reply []byte) {
p.Size = uint32(len(reply))
p.Data = make([]byte, p.Size)
copy(p.Data[:p.Size], reply)
p.ResultCode = code
p.ArgLen = 0
}
// GetUniqueLogId returns the unique log ID.
func (p *Packet) GetUniqueLogId() (m string) {
m = fmt.Sprintf("Req(%v)_Partition(%v)_", p.ReqID, p.PartitionID)
if p.ExtentType == TinyExtentType && p.Opcode == OpMarkDelete && len(p.Data) > 0 {
ext := new(TinyExtentDeleteRecord)
err := json.Unmarshal(p.Data, ext)
if err == nil {
m += fmt.Sprintf("Extent(%v)_ExtentOffset(%v)_TinyDeleteFileOffset(%v)_Size(%v)_Opcode(%v)_ResultCode(%v)",
ext.ExtentId, ext.ExtentOffset, ext.TinyDeleteFileOffset, ext.Size, p.Opcode, p.ResultCode)
return m
}
} else if p.Opcode == OpReadTinyDelete {
m += fmt.Sprintf("Opcode(%v)_ResultCode(%v)", p.GetOpMsg(), p.GetResultMsg())
return m
} else if p.Opcode == OpNotifyReplicasToRepair {
m += fmt.Sprintf("Opcode(%v)_ResultCode(%v)", p.GetOpMsg(), p.GetResultMsg())
return m
}
m = fmt.Sprintf("Req(%v)_Partition(%v)_Extent(%v)_ExtentOffset(%v)_KernelOffset(%v)_"+
"Size(%v)_Opcode(%v)_ResultCode(%v)_CRC(%v)",
p.ReqID, p.PartitionID, p.ExtentID, p.ExtentOffset,
p.KernelOffset, p.Size, p.GetOpMsg(), p.GetResultMsg(), p.CRC)
return
}
// IsForwardPkt returns if the packet is the forward packet (a packet that will be forwarded to the followers).
func (p *Packet) IsForwardPkt() bool {
return p.RemainingFollowers > 0
}
// LogMessage logs the given message.
func (p *Packet) LogMessage(action, remote string, start int64, err error) (m string) {
if err == nil {
m = fmt.Sprintf("id[%v] op[%v] remote[%v] "+
" cost[%v] transite[%v] nodes[%v]",
p.GetUniqueLogId(), action, remote,
(time.Now().UnixNano()-start)/1e6, p.IsForwardPkt(), p.RemainingFollowers)
} else {
m = fmt.Sprintf("id[%v] op[%v] remote[%v]"+
", err[%v] transite[%v] nodes[%v]", p.GetUniqueLogId(), action,
remote, err.Error(), p.IsForwardPkt(), p.RemainingFollowers)
}
return
}
// ShallRetry returns if we should retry the packet.
func (p *Packet) ShouldRetry() bool {
return p.ResultCode == OpAgain || p.ResultCode == OpErr
}

22
proto/status.go Normal file
View File

@ -0,0 +1,22 @@
// Copyright 2018 The Chubao 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 proto
// The following defines the status of a disk or a partition.
const (
ReadOnly = 1
ReadWrite = 2
Unavailable = -1
)

2
repl/repl_protocol.go
View File

@ -326,8 +326,6 @@ func (rp *ReplProtocol) writeResponse(reply *Packet) {
// execute the post-processing function
rp.postFunc(reply)
if !reply.NeedReply {
log.LogDebugf(reply.LogMessage(ActionWriteToClient,
rp.sourceConn.RemoteAddr().String(), reply.StartT, err))
return
}