// Copyright 2018 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 datanode import ( "encoding/json" "fmt" "hash/crc32" "sync/atomic" "github.com/cubefs/cubefs/datanode/repl" "github.com/cubefs/cubefs/datanode/storage" "github.com/cubefs/cubefs/proto" "github.com/cubefs/cubefs/util/log" ) func (s *DataNode) Prepare(p *repl.Packet) (err error) { defer func() { p.SetPacketHasPrepare() if err != nil { p.PackErrorBody(repl.ActionPreparePkt, err.Error()) } else { p.AfterPre = true } }() // service not start and not urgent req. if !s.HasStarted() && !p.IsUrgentLeaderReq() { log.LogDebugf("prepare: datanode still not start, op %s", p.GetOpMsg()) return fmt.Errorf("datanode still not started") } if p.IsMasterCommand() { return } atomic.AddUint64(&s.metricsCnt, 1) if !s.shallDegrade() { p.BeforeTp(s.clusterID) p.UnsetDegrade() } else { p.SetDegrade() } err = s.checkStoreMode(p) if err != nil { return } if err = s.checkCrc(p); err != nil { return } if err = s.checkPartition(p); err != nil { return } // For certain packet, we need to add some additional extent information. if err = s.checkPacketAndPrepare(p); err != nil { return } return } func (s *DataNode) checkStoreMode(p *repl.Packet) (err error) { if proto.IsTinyExtentType(p.ExtentType) || proto.IsNormalExtentType(p.ExtentType) { return } log.LogErrorf("action[checkStoreMode] dp [%v] reqId [%v] extent type %v", p.PartitionID, p.ReqID, p.ExtentType) return ErrIncorrectStoreType } func (s *DataNode) checkCrc(p *repl.Packet) (err error) { if !p.IsNormalWriteOperation() { return } crc := crc32.ChecksumIEEE(p.Data[:p.Size]) if crc != p.CRC { return storage.CrcMismatchError } return } func (s *DataNode) checkPartition(p *repl.Packet) (err error) { dp := s.space.Partition(p.PartitionID) if dp == nil { // err = proto.ErrDataPartitionNotExists err = fmt.Errorf("%v %v", proto.ErrDataPartitionNotExists.Error(), p.PartitionID) return } p.Object = dp if p.IsNormalWriteOperation() || p.IsCreateExtentOperation() { if dp.Available() <= 0 { log.LogErrorf("[checkPartition] dp(%v) disk no space available(%v) can write(%v)", dp.partitionID, dp.Available(), dp.disk.CanWrite()) err = storage.NoSpaceError return } } if p.IsNormalWriteOperation() || p.IsRandomWrite() { dp.disk.allocCheckLimit(proto.FlowWriteType, uint32(p.Size)) dp.disk.allocCheckLimit(proto.IopsWriteType, 1) } return } func (s *DataNode) checkPacketAndPrepare(p *repl.Packet) error { partition := p.Object.(*DataPartition) store := p.Object.(*DataPartition).ExtentStore() var ( extentID uint64 err error ) if log.EnableDebug() { log.LogDebugf("action[prepare.checkPacketAndPrepare] pack opcode (%v) p.IsLeaderPacket(%v) p (%v)", p.Opcode, p.IsLeaderPacket(), p) } if p.IsRandomWrite() || p.IsSnapshotModWriteAppendOperation() || p.IsNormalWriteOperation() { if err = partition.CheckWriteVer(p); err != nil { return err } } if p.IsLeaderPacket() && proto.IsTinyExtentType(p.ExtentType) && p.IsNormalWriteOperation() { extentID, err = store.GetAvailableTinyExtent() if err != nil { return fmt.Errorf("checkPacketAndPrepare partition %v GetAvailableTinyExtent error %v", p.PartitionID, err.Error()) } p.ExtentID = extentID p.ExtentOffset, err = store.GetTinyExtentOffset(extentID) if err != nil { return fmt.Errorf("checkPacketAndPrepare partition %v %v GetTinyExtentOffset error %v", p.PartitionID, extentID, err.Error()) } } else if p.IsLeaderPacket() && p.IsSnapshotModWriteAppendOperation() { if proto.IsTinyExtentType(p.ExtentType) { extentID, err = store.GetAvailableTinyExtent() if err != nil { log.LogErrorf("err %v", err) return fmt.Errorf("checkPacketAndPrepare partition %v GetAvailableTinyExtent error %v", p.PartitionID, err.Error()) } p.ExtentID = extentID p.ExtentOffset, err = store.GetTinyExtentOffset(p.ExtentID) if err != nil { err = fmt.Errorf("checkPacketAndPrepare partition %v %v GetTinyExtentOffset error %v", p.PartitionID, extentID, err.Error()) log.LogErrorf("err %v", err) } log.LogDebugf("action[prepare.checkPacketAndPrepare] dp %v append randomWrite p.ExtentOffset %v Kernel(file)Offset %v", p.PartitionID, p.ExtentOffset, p.KernelOffset) return err } p.ExtentOffset, err = store.GetExtentSnapshotModOffset(p.ExtentID, p.Size) log.LogDebugf("action[prepare.checkPacketAndPrepare] pack (%v) partition %v %v", p, p.PartitionID, extentID) if err != nil { return fmt.Errorf("checkPacketAndPrepare partition %v %v GetSnapshotModExtentOffset error %v", p.PartitionID, extentID, err.Error()) } } else if p.IsLeaderPacket() && p.IsCreateExtentOperation() { if partition.isNormalType() && partition.GetExtentCount() >= storage.MaxExtentCount*3 { log.LogErrorf("[checkPacketAndPrepare] partition %v has reached maxExtentId", p.PartitionID) return storage.ReachMaxExtentsCountError } p.ExtentID, err = store.NextExtentID() if err != nil { return fmt.Errorf("checkPacketAndPrepare partition %v allocCheckLimit NextExtentId error %v", p.PartitionID, err) } } else if p.IsLeaderPacket() && ((p.IsMarkDeleteExtentOperation() && proto.IsTinyExtentType(p.ExtentType)) || (p.IsMarkSplitExtentOperation() && !proto.IsTinyExtentType(p.ExtentType))) { log.LogDebugf("checkPacketAndPrepare. packet opCode %v p.ExtentType %v", p.Opcode, p.ExtentType) record := new(proto.TinyExtentDeleteRecord) if err := json.Unmarshal(p.Data[:p.Size], record); err != nil { return fmt.Errorf("checkPacketAndPrepare failed %v", err.Error()) } p.Data, _ = json.Marshal(record) p.Size = uint32(len(p.Data)) } if (p.IsCreateExtentOperation() || p.IsNormalWriteOperation()) && p.ExtentID == 0 { return fmt.Errorf("checkPacketAndPrepare partition %v invalid extent id. ", p.PartitionID) } p.OrgBuffer = p.Data return nil }