add when datanode disk error ,then ump warn,and change disk error to -1

Signed-off-by: awzhgw <guowl18702995996@gmail.com>

datanode/disk.go

@@ -27,6 +27,7 @@ import (
 	"time"
 
 	"github.com/chubaofs/chubaofs/proto"
+	"github.com/chubaofs/chubaofs/util/exporter"
 	"github.com/chubaofs/chubaofs/util/log"
 )
 
@@ -162,6 +163,8 @@ func (d *Disk) updateSpaceInfo() (err error) {
 	currErrs := d.ReadErrCnt + d.WriteErrCnt
 	if currErrs >= uint64(d.MaxErrCnt) {
 		d.Status = proto.Unavailable
+		exporter.NewAlarm(fmt.Sprintf("disk path %v error on %v", d.Path, LocalIP))
+		d.ForceExitRaftStore()
 	} else if d.Available <= 0 {
 		d.Status = proto.ReadOnly
 	} else {
@@ -206,6 +209,15 @@ func (d *Disk) ForceLoadPartitionHeader() {
 	}
 }
 
+func (d *Disk) ForceExitRaftStore(){
+	partitionList := d.DataPartitionList()
+	for _, partitionID := range partitionList {
+		partition := d.GetDataPartition(partitionID)
+		partition.partitionStatus=proto.Unavailable
+		partition.stopRaft()
+	}
+}
+
 // DataPartitionList returns a list of the data partitions
 func (d *Disk) DataPartitionList() (partitionIDs []uint64) {
 	d.Lock()

datanode/partition.go

@@ -32,6 +32,7 @@ import (
 	"github.com/chubaofs/chubaofs/repl"
 	"github.com/chubaofs/chubaofs/storage"
 	"github.com/chubaofs/chubaofs/util/errors"
+	"github.com/chubaofs/chubaofs/util/exporter"
 	"github.com/chubaofs/chubaofs/util/log"
 	raftProto "github.com/tiglabs/raft/proto"
 	"hash/crc32"
@@ -425,6 +426,23 @@ func (dp *DataPartition) ExtentStore() *storage.ExtentStore {
 	return dp.extentStore
 }
 
+func (dp *DataPartition) checkIsDiskError(err error) {
+	if err == nil {
+		return
+	}
+	if IsDiskErr(err.Error()) {
+		mesg := fmt.Sprintf("disk path %v error on %v", dp.Path, LocalIP)
+		exporter.NewAlarm(mesg)
+		log.LogErrorf(mesg)
+		dp.stopRaft()
+		dp.disk.incReadErrCnt()
+		dp.disk.incWriteErrCnt()
+		dp.disk.Status = proto.Unavailable
+		dp.statusUpdate()
+		dp.disk.ForceExitRaftStore()
+	}
+}
+
 // String returns the string format of the data partition information.
 func (dp *DataPartition) String() (m string) {
 	return fmt.Sprintf(DataPartitionPrefix+"_%v_%v", dp.partitionID, dp.partitionSize)

datanode/partition_op_by_raft.go

@@ -217,6 +217,7 @@ func (dp *DataPartition) ApplyRandomWrite(msg *RaftCmdItem, raftApplyID uint64)
 		raftApplyID, dp.partitionID, opItem.extentID, opItem.offset, opItem.size)
 	for i := 0; i < maxRetryCounts; i++ {
 		err = dp.ExtentStore().Write(opItem.extentID, opItem.offset, opItem.size, opItem.data, opItem.crc, NotUpdateSize, msg.Op == opRandomSyncWrite)
+		dp.checkIsDiskError(err)
 		if err != nil {
 			if dp.checkWriteErrs(err.Error()) {
 				log.LogErrorf("[ApplyRandomWrite] ApplyID(%v) Partition(%v)_Extent(%v)_ExtentOffset(%v)_Size(%v) ignore error[%v]",

datanode/server.go

@@ -53,7 +53,7 @@ const (
 	DefaultRackName         = "cfs_rack1"
 	DefaultRaftDir          = "raft"
 	DefaultRaftLogsToRetain = 2000 // Count of raft logs per data partition
-	DefaultDiskMaxErr       = 20
+	DefaultDiskMaxErr       = 1
 	DefaultDiskRetain       = 20 * util.GB // GB
 )
 

datanode/wrap_operator.go

@@ -362,6 +362,7 @@ func (s *DataNode) handleWritePacket(p *repl.Packet) {
 	store := partition.ExtentStore()
 	if p.Size <= util.BlockSize {
 		err = store.Write(p.ExtentID, p.ExtentOffset, int64(p.Size), p.Data, p.CRC, UpdateSize, p.Opcode == proto.OpSyncWrite)
+		partition.checkIsDiskError(err)
 	} else {
 		size := p.Size
 		offset := 0
@@ -373,6 +374,7 @@ func (s *DataNode) handleWritePacket(p *repl.Packet) {
 			data := p.Data[offset : offset+currSize]
 			crc := crc32.ChecksumIEEE(data)
 			err = store.Write(p.ExtentID, p.ExtentOffset+int64(offset), int64(currSize), data, crc, UpdateSize, p.Opcode == proto.OpSyncWrite)
+			partition.checkIsDiskError(err)
 			if err != nil {
 				break
 			}
@@ -467,6 +469,7 @@ func (s *DataNode) handleExtentRepaiReadPacket(p *repl.Packet, connect net.Conn,
 		p.Size = uint32(currReadSize)
 		p.ExtentOffset = offset
 		reply.CRC, err = store.Read(reply.ExtentID, offset, int64(currReadSize), reply.Data, isRepairRead)
+		partition.checkIsDiskError(err)
 		p.CRC = reply.CRC
 		tpObject.Set()
 		if err != nil {

metanode/partition.go

@@ -298,6 +298,7 @@ func (mp *metaPartition) stopRaft() {
 	if mp.raftPartition != nil {
 		// TODO Unhandled errors
 		mp.raftPartition.Stop()
+		mp.raftPartition=nil
 	}
 	return
 }

util/exporter/alarm.go

@@ -16,6 +16,7 @@ package exporter
 
 import (
 	"fmt"
+	"github.com/chubaofs/chubaofs/util/ump"
 	"sync"
 )
 
@@ -39,12 +40,13 @@ type Alarm struct {
 	Counter
 }
 
-func NewAlarm(name string) (a *Alarm) {
+func NewAlarm(key string) (a *Alarm) {
 	if !enabled {
+		ump.Alarm(fmt.Sprintf("%v_%v_warning", clustername, modulename), key)
 		return
 	}
 	a = AlarmPool.Get().(*Alarm)
-	a.name = metricsName(fmt.Sprintf("%s_alarm", name))
+	a.name = metricsName(fmt.Sprintf("%s_alarm", key))
 	a.Add(1)
 	return
 }