mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 02:00:56 +00:00
feat(meta): support back-end audit log
close: #2625 Signed-off-by: NaturalSelect <2145973003@qq.com>
This commit is contained in:
NaturalSelect
Victor1319
parent
804fd4780d
commit
e839eae5a7
11
cmd/cmd.go
11
cmd/cmd.go
@ -29,6 +29,7 @@ import (
|
||||
"strings"
|
||||
"syscall"
|
||||
|
||||
"github.com/cubefs/cubefs/util/auditlog"
|
||||
"github.com/cubefs/cubefs/util/errors"
|
||||
sysutil "github.com/cubefs/cubefs/util/sys"
|
||||
|
||||
@ -169,7 +170,6 @@ func main() {
|
||||
log.LogErrorf("logLeftSpaceLimit is not a legal int value: %v", err.Error())
|
||||
logLeftSpaceLimit = log.DefaultLogLeftSpaceLimit
|
||||
}
|
||||
|
||||
// Init server instance with specified role configuration.
|
||||
var (
|
||||
server common.Server
|
||||
@ -232,6 +232,15 @@ func main() {
|
||||
}
|
||||
defer log.LogFlush()
|
||||
|
||||
_, err = auditlog.InitAudit(logDir, module, auditlog.DefaultAuditLogSize)
|
||||
if err != nil {
|
||||
err = errors.NewErrorf("Fatal: failed to init audit log - %v", err)
|
||||
fmt.Println(err)
|
||||
daemonize.SignalOutcome(err)
|
||||
os.Exit(1)
|
||||
}
|
||||
defer auditlog.StopAudit()
|
||||
|
||||
if *redirectSTD {
|
||||
// Init output file
|
||||
outputFilePath := path.Join(logDir, module, LoggerOutput)
|
||||
|
||||
@ -212,7 +212,7 @@ func (m *metadataManager) opCreateInode(conn net.Conn, p *Packet,
|
||||
return
|
||||
}
|
||||
|
||||
err = mp.CreateInode(req, p)
|
||||
err = mp.CreateInode(req, p, remoteAddr)
|
||||
// reply the operation result to the client through TCP
|
||||
m.respondToClientWithVer(conn, p)
|
||||
log.LogDebugf("%s [opCreateInode] req: %d - %v, resp: %v, body: %s",
|
||||
@ -242,7 +242,7 @@ func (m *metadataManager) opQuotaCreateInode(conn net.Conn, p *Packet, remoteAdd
|
||||
err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
|
||||
return
|
||||
}
|
||||
err = mp.QuotaCreateInode(req, p)
|
||||
err = mp.QuotaCreateInode(req, p, remoteAddr)
|
||||
// reply the operation result to the client through TCP
|
||||
m.respondToClient(conn, p)
|
||||
log.LogDebugf("%s [opQuotaCreateInode] req: %d - %v, resp: %v, body: %s",
|
||||
@ -272,7 +272,7 @@ func (m *metadataManager) opTxMetaLinkInode(conn net.Conn, p *Packet, remoteAddr
|
||||
err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
|
||||
return
|
||||
}
|
||||
err = mp.TxCreateInodeLink(req, p)
|
||||
err = mp.TxCreateInodeLink(req, p, remoteAddr)
|
||||
m.respondToClient(conn, p)
|
||||
log.LogDebugf("%s [opTxMetaLinkInode] req: %d - %v, resp: %v, body: %s",
|
||||
remoteAddr, p.GetReqID(), req, p.GetResultMsg(), p.Data)
|
||||
@ -302,7 +302,7 @@ func (m *metadataManager) opMetaLinkInode(conn net.Conn, p *Packet,
|
||||
err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
|
||||
return
|
||||
}
|
||||
err = mp.CreateInodeLink(req, p)
|
||||
err = mp.CreateInodeLink(req, p, remoteAddr)
|
||||
m.respondToClientWithVer(conn, p)
|
||||
log.LogDebugf("%s [opMetaLinkInode] req: %d - %v, resp: %v, body: %s",
|
||||
remoteAddr, p.GetReqID(), req, p.GetResultMsg(), p.Data)
|
||||
@ -352,7 +352,7 @@ func (m *metadataManager) opTxCreateDentry(conn net.Conn, p *Packet,
|
||||
err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
|
||||
return
|
||||
}
|
||||
err = mp.TxCreateDentry(req, p)
|
||||
err = mp.TxCreateDentry(req, p, remoteAddr)
|
||||
m.respondToClient(conn, p)
|
||||
|
||||
log.LogDebugf("%s [opTxCreateDentry] req: %d - %v, resp: %v, body: %s",
|
||||
@ -504,7 +504,7 @@ func (m *metadataManager) opTxCommit(conn net.Conn, p *Packet,
|
||||
return
|
||||
}
|
||||
|
||||
err = mp.TxCommit(req, p)
|
||||
err = mp.TxCommit(req, p, remoteAddr)
|
||||
m.respondToClient(conn, p)
|
||||
|
||||
log.LogDebugf("%s [opTxCommit] req: %d - %v, resp: %v, body: %s",
|
||||
@ -534,7 +534,7 @@ func (m *metadataManager) opTxRollback(conn net.Conn, p *Packet,
|
||||
return
|
||||
}
|
||||
|
||||
err = mp.TxRollback(req, p)
|
||||
err = mp.TxRollback(req, p, remoteAddr)
|
||||
m.respondToClient(conn, p)
|
||||
|
||||
log.LogDebugf("%s [opTxRollback] req: %d - %v, resp: %v, body: %s",
|
||||
@ -566,7 +566,7 @@ func (m *metadataManager) opCreateDentry(conn net.Conn, p *Packet,
|
||||
err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
|
||||
return
|
||||
}
|
||||
err = mp.CreateDentry(req, p)
|
||||
err = mp.CreateDentry(req, p, remoteAddr)
|
||||
m.respondToClient(conn, p)
|
||||
|
||||
log.LogDebugf("%s [opCreateDentry] req: %d - %v, resp: %v, body: %s",
|
||||
@ -597,7 +597,7 @@ func (m *metadataManager) opQuotaCreateDentry(conn net.Conn, p *Packet,
|
||||
err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
|
||||
return
|
||||
}
|
||||
err = mp.QuotaCreateDentry(req, p)
|
||||
err = mp.QuotaCreateDentry(req, p, remoteAddr)
|
||||
m.respondToClient(conn, p)
|
||||
|
||||
log.LogDebugf("%s [opQuotaCreateDentry] req: %d - %v, resp: %v, body: %s",
|
||||
@ -627,7 +627,7 @@ func (m *metadataManager) opTxDeleteDentry(conn net.Conn, p *Packet,
|
||||
if !m.serveProxy(conn, mp, p) {
|
||||
return
|
||||
}
|
||||
err = mp.TxDeleteDentry(req, p)
|
||||
err = mp.TxDeleteDentry(req, p, remoteAddr)
|
||||
m.respondToClient(conn, p)
|
||||
if log.EnableDebug() {
|
||||
log.LogDebugf("%s [opTxDeleteDentry] req: %d - %v, resp: %v, body: %s",
|
||||
@ -662,7 +662,7 @@ func (m *metadataManager) opDeleteDentry(conn net.Conn, p *Packet,
|
||||
return
|
||||
}
|
||||
|
||||
err = mp.DeleteDentry(req, p)
|
||||
err = mp.DeleteDentry(req, p, remoteAddr)
|
||||
m.respondToClient(conn, p)
|
||||
log.LogDebugf("%s [opDeleteDentry] req: %d - %v, resp: %v, body: %s",
|
||||
remoteAddr, p.GetReqID(), req, p.GetResultMsg(), p.Data)
|
||||
@ -695,7 +695,7 @@ func (m *metadataManager) opBatchDeleteDentry(conn net.Conn, p *Packet,
|
||||
return
|
||||
}
|
||||
|
||||
err = mp.DeleteDentryBatch(req, p)
|
||||
err = mp.DeleteDentryBatch(req, p, remoteAddr)
|
||||
|
||||
m.respondToClientWithVer(conn, p)
|
||||
log.LogDebugf("%s [opDeleteDentry] req: %d - %v, resp: %v, body: %s",
|
||||
@ -726,7 +726,7 @@ func (m *metadataManager) opTxUpdateDentry(conn net.Conn, p *Packet, remoteAddr
|
||||
err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
|
||||
return
|
||||
}
|
||||
err = mp.TxUpdateDentry(req, p)
|
||||
err = mp.TxUpdateDentry(req, p, remoteAddr)
|
||||
|
||||
m.respondToClientWithVer(conn, p)
|
||||
log.LogDebugf("%s [opTxUpdateDentry] req: %d - %v; resp: %v, body: %s",
|
||||
@ -759,7 +759,7 @@ func (m *metadataManager) opUpdateDentry(conn net.Conn, p *Packet,
|
||||
return
|
||||
}
|
||||
|
||||
err = mp.UpdateDentry(req, p)
|
||||
err = mp.UpdateDentry(req, p, remoteAddr)
|
||||
|
||||
m.respondToClientWithVer(conn, p)
|
||||
log.LogDebugf("%s [opUpdateDentry] req: %d - %v; resp: %v, body: %s",
|
||||
@ -791,7 +791,7 @@ func (m *metadataManager) opTxMetaUnlinkInode(conn net.Conn, p *Packet, remoteAd
|
||||
err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
|
||||
return
|
||||
}
|
||||
err = mp.TxUnlinkInode(req, p)
|
||||
err = mp.TxUnlinkInode(req, p, remoteAddr)
|
||||
|
||||
m.respondToClientWithVer(conn, p)
|
||||
log.LogDebugf("%s [opDeleteInode] req: %d - %v, resp: %v, body: %s",
|
||||
@ -822,7 +822,7 @@ func (m *metadataManager) opMetaUnlinkInode(conn net.Conn, p *Packet,
|
||||
err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
|
||||
return
|
||||
}
|
||||
err = mp.UnlinkInode(req, p)
|
||||
err = mp.UnlinkInode(req, p, remoteAddr)
|
||||
m.respondToClientWithVer(conn, p)
|
||||
log.LogDebugf("%s [opDeleteInode] req: %d - %v, resp: %v, body: %s",
|
||||
remoteAddr, p.GetReqID(), req, p.GetResultMsg(), p.Data)
|
||||
@ -852,7 +852,7 @@ func (m *metadataManager) opMetaBatchUnlinkInode(conn net.Conn, p *Packet,
|
||||
err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
|
||||
return
|
||||
}
|
||||
err = mp.UnlinkInodeBatch(req, p)
|
||||
err = mp.UnlinkInodeBatch(req, p, remoteAddr)
|
||||
m.respondToClientWithVer(conn, p)
|
||||
log.LogDebugf("%s [opDeleteInode] req: %d - %v, resp: %v, body: %s",
|
||||
remoteAddr, p.GetReqID(), req, p.GetResultMsg(), p.Data)
|
||||
@ -1007,7 +1007,7 @@ func (m *metadataManager) opBatchMetaEvictInode(conn net.Conn, p *Packet,
|
||||
err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
|
||||
return
|
||||
}
|
||||
if err = mp.EvictInodeBatch(req, p); err != nil {
|
||||
if err = mp.EvictInodeBatch(req, p, remoteAddr); err != nil {
|
||||
err = errors.NewErrorf("[%v] req: %v, resp: %v", p.GetOpMsgWithReqAndResult(), req, err.Error())
|
||||
}
|
||||
m.respondToClientWithVer(conn, p)
|
||||
@ -1039,7 +1039,7 @@ func (m *metadataManager) opMetaEvictInode(conn net.Conn, p *Packet,
|
||||
err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
|
||||
return
|
||||
}
|
||||
if err = mp.EvictInode(req, p); err != nil {
|
||||
if err = mp.EvictInode(req, p, remoteAddr); err != nil {
|
||||
err = errors.NewErrorf("[%v] req: %v, resp: %v", p.GetOpMsgWithReqAndResult(), req, err.Error())
|
||||
}
|
||||
m.respondToClientWithVer(conn, p)
|
||||
@ -1286,7 +1286,7 @@ func (m *metadataManager) opMetaExtentsTruncate(conn net.Conn, p *Packet,
|
||||
err = errors.NewErrorf("[%v],req[%v],err[%v]", p.GetOpMsgWithReqAndResult(), req, string(p.Data))
|
||||
return
|
||||
}
|
||||
mp.ExtentsTruncate(req, p)
|
||||
mp.ExtentsTruncate(req, p, remoteAddr)
|
||||
m.respondToClientWithVer(conn, p)
|
||||
log.LogDebugf("%s [OpMetaTruncate] req: %d - %v, resp body: %v, "+
|
||||
"resp body: %s", remoteAddr, p.GetReqID(), req, p.GetResultMsg(), p.Data)
|
||||
@ -1699,7 +1699,7 @@ func (m *metadataManager) opMetaDeleteInode(conn net.Conn, p *Packet,
|
||||
if !m.serveProxy(conn, mp, p) {
|
||||
return
|
||||
}
|
||||
err = mp.DeleteInode(req, p)
|
||||
err = mp.DeleteInode(req, p, remoteAddr)
|
||||
_ = m.respondToClientWithVer(conn, p)
|
||||
log.LogDebugf("%s [opMetaDeleteInode] req: %d - %v, resp: %v, body: %s",
|
||||
remoteAddr, p.GetReqID(), req, p.GetResultMsg(), p.Data)
|
||||
@ -1726,7 +1726,7 @@ func (m *metadataManager) opMetaBatchDeleteInode(conn net.Conn, p *Packet,
|
||||
if !m.serveProxy(conn, mp, p) {
|
||||
return
|
||||
}
|
||||
err = mp.DeleteInodeBatch(req, p)
|
||||
err = mp.DeleteInodeBatch(req, p, remoteAddr)
|
||||
log.LogDebugf("%s [opMetaDeleteInode] req: %d - %v, resp: %v, body: %s",
|
||||
remoteAddr, p.GetReqID(), req, p.GetResultMsg(), p.Data)
|
||||
|
||||
@ -2170,7 +2170,7 @@ func (m *metadataManager) opTxCreateInode(conn net.Conn, p *Packet,
|
||||
return
|
||||
}
|
||||
|
||||
err = mp.TxCreateInode(req, p)
|
||||
err = mp.TxCreateInode(req, p, remoteAddr)
|
||||
m.respondToClient(conn, p)
|
||||
log.LogDebugf("%s [opTxCreateInode] req: %d - %v, resp: %v, body: %s",
|
||||
remoteAddr, p.GetReqID(), req, p.GetResultMsg(), p.Data)
|
||||
|
||||
@ -1,7 +1,28 @@
|
||||
// Copyright 2023 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 metanode
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math"
|
||||
"os"
|
||||
"reflect"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
raftstoremock "github.com/cubefs/cubefs/metanode/mocktest/raftstore"
|
||||
"github.com/cubefs/cubefs/proto"
|
||||
"github.com/cubefs/cubefs/util"
|
||||
@ -9,12 +30,6 @@ import (
|
||||
"github.com/cubefs/cubefs/util/log"
|
||||
"github.com/golang/mock/gomock"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"math"
|
||||
"os"
|
||||
"reflect"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
var partitionId uint64 = 10
|
||||
@ -1443,10 +1458,10 @@ func TestXAttrOperation(t *testing.T) {
|
||||
func TestUpdateDenty(t *testing.T) {
|
||||
newMpWithMock(t)
|
||||
testCreateInode(nil, DirModeType)
|
||||
err := mp.CreateDentry(&CreateDentryReq{Name: "testfile", ParentID: 1, VerSeq: 0, Inode: 1000}, &Packet{})
|
||||
err := mp.CreateDentry(&CreateDentryReq{Name: "testfile", ParentID: 1, VerSeq: 0, Inode: 1000}, &Packet{}, localAddrForAudit)
|
||||
assert.True(t, err == nil)
|
||||
testCreateVer()
|
||||
mp.UpdateDentry(&UpdateDentryReq{Name: "testfile", ParentID: 1, Inode: 2000}, &Packet{})
|
||||
mp.UpdateDentry(&UpdateDentryReq{Name: "testfile", ParentID: 1, Inode: 2000}, &Packet{}, localAddrForAudit)
|
||||
den := &Dentry{Name: "testfile", ParentId: 1}
|
||||
den.setVerSeq(math.MaxUint64)
|
||||
denRsp, status := mp.getDentry(den)
|
||||
|
||||
@ -43,6 +43,10 @@ import (
|
||||
"github.com/cubefs/cubefs/util/timeutil"
|
||||
)
|
||||
|
||||
// NOTE: if the operation is invoked by local machine
|
||||
// the remote addr is "127.0.0.1"
|
||||
const localAddrForAudit = "127.0.0.1"
|
||||
|
||||
var (
|
||||
ErrIllegalHeartbeatAddress = errors.New("illegal heartbeat address")
|
||||
ErrIllegalReplicateAddress = errors.New("illegal replicate address")
|
||||
@ -126,25 +130,25 @@ func (c *MetaPartitionConfig) sortPeers() {
|
||||
|
||||
// OpInode defines the interface for the inode operations.
|
||||
type OpInode interface {
|
||||
CreateInode(req *CreateInoReq, p *Packet) (err error)
|
||||
UnlinkInode(req *UnlinkInoReq, p *Packet) (err error)
|
||||
UnlinkInodeBatch(req *BatchUnlinkInoReq, p *Packet) (err error)
|
||||
CreateInode(req *CreateInoReq, p *Packet, remoteAddr string) (err error)
|
||||
UnlinkInode(req *UnlinkInoReq, p *Packet, remoteAddr string) (err error)
|
||||
UnlinkInodeBatch(req *BatchUnlinkInoReq, p *Packet, remoteAddr string) (err error)
|
||||
InodeGet(req *InodeGetReq, p *Packet) (err error)
|
||||
InodeGetSplitEk(req *InodeGetSplitReq, p *Packet) (err error)
|
||||
InodeGetBatch(req *InodeGetReqBatch, p *Packet) (err error)
|
||||
CreateInodeLink(req *LinkInodeReq, p *Packet) (err error)
|
||||
EvictInode(req *EvictInodeReq, p *Packet) (err error)
|
||||
EvictInodeBatch(req *BatchEvictInodeReq, p *Packet) (err error)
|
||||
CreateInodeLink(req *LinkInodeReq, p *Packet, remoteAddr string) (err error)
|
||||
EvictInode(req *EvictInodeReq, p *Packet, remoteAddr string) (err error)
|
||||
EvictInodeBatch(req *BatchEvictInodeReq, p *Packet, remoteAddr string) (err error)
|
||||
SetAttr(req *SetattrRequest, reqData []byte, p *Packet) (err error)
|
||||
GetInodeTree() *BTree
|
||||
GetInodeTreeLen() int
|
||||
DeleteInode(req *proto.DeleteInodeRequest, p *Packet) (err error)
|
||||
DeleteInodeBatch(req *proto.DeleteInodeBatchRequest, p *Packet) (err error)
|
||||
DeleteInode(req *proto.DeleteInodeRequest, p *Packet, remoteAddr string) (err error)
|
||||
DeleteInodeBatch(req *proto.DeleteInodeBatchRequest, p *Packet, remoteAddr string) (err error)
|
||||
ClearInodeCache(req *proto.ClearInodeCacheRequest, p *Packet) (err error)
|
||||
TxCreateInode(req *proto.TxCreateInodeRequest, p *Packet) (err error)
|
||||
TxUnlinkInode(req *proto.TxUnlinkInodeRequest, p *Packet) (err error)
|
||||
TxCreateInodeLink(req *proto.TxLinkInodeRequest, p *Packet) (err error)
|
||||
QuotaCreateInode(req *proto.QuotaCreateInodeRequest, p *Packet) (err error)
|
||||
TxCreateInode(req *proto.TxCreateInodeRequest, p *Packet, remoteAddr string) (err error)
|
||||
TxUnlinkInode(req *proto.TxUnlinkInodeRequest, p *Packet, remoteAddr string) (err error)
|
||||
TxCreateInodeLink(req *proto.TxLinkInodeRequest, p *Packet, remoteAddr string) (err error)
|
||||
QuotaCreateInode(req *proto.QuotaCreateInodeRequest, p *Packet, remoteAddr string) (err error)
|
||||
}
|
||||
|
||||
type OpExtend interface {
|
||||
@ -160,28 +164,28 @@ type OpExtend interface {
|
||||
|
||||
// OpDentry defines the interface for the dentry operations.
|
||||
type OpDentry interface {
|
||||
CreateDentry(req *CreateDentryReq, p *Packet) (err error)
|
||||
DeleteDentry(req *DeleteDentryReq, p *Packet) (err error)
|
||||
DeleteDentryBatch(req *BatchDeleteDentryReq, p *Packet) (err error)
|
||||
UpdateDentry(req *UpdateDentryReq, p *Packet) (err error)
|
||||
CreateDentry(req *CreateDentryReq, p *Packet, remoteAddr string) (err error)
|
||||
DeleteDentry(req *DeleteDentryReq, p *Packet, remoteAddr string) (err error)
|
||||
DeleteDentryBatch(req *BatchDeleteDentryReq, p *Packet, remoteAddr string) (err error)
|
||||
UpdateDentry(req *UpdateDentryReq, p *Packet, remoteAddr string) (err error)
|
||||
ReadDir(req *ReadDirReq, p *Packet) (err error)
|
||||
ReadDirLimit(req *ReadDirLimitReq, p *Packet) (err error)
|
||||
ReadDirOnly(req *ReadDirOnlyReq, p *Packet) (err error)
|
||||
Lookup(req *LookupReq, p *Packet) (err error)
|
||||
GetDentryTree() *BTree
|
||||
GetDentryTreeLen() int
|
||||
TxCreateDentry(req *proto.TxCreateDentryRequest, p *Packet) (err error)
|
||||
TxDeleteDentry(req *proto.TxDeleteDentryRequest, p *Packet) (err error)
|
||||
TxUpdateDentry(req *proto.TxUpdateDentryRequest, p *Packet) (err error)
|
||||
QuotaCreateDentry(req *proto.QuotaCreateDentryRequest, p *Packet) (err error)
|
||||
TxCreateDentry(req *proto.TxCreateDentryRequest, p *Packet, remoteAddr string) (err error)
|
||||
TxDeleteDentry(req *proto.TxDeleteDentryRequest, p *Packet, remoteAddr string) (err error)
|
||||
TxUpdateDentry(req *proto.TxUpdateDentryRequest, p *Packet, remoteAddr string) (err error)
|
||||
QuotaCreateDentry(req *proto.QuotaCreateDentryRequest, p *Packet, remoteAddr string) (err error)
|
||||
}
|
||||
|
||||
type OpTransaction interface {
|
||||
TxCreate(req *proto.TxCreateRequest, p *Packet) (err error)
|
||||
TxCommitRM(req *proto.TxApplyRMRequest, p *Packet) error
|
||||
TxRollbackRM(req *proto.TxApplyRMRequest, p *Packet) error
|
||||
TxCommit(req *proto.TxApplyRequest, p *Packet) (err error)
|
||||
TxRollback(req *proto.TxApplyRequest, p *Packet) (err error)
|
||||
TxCommit(req *proto.TxApplyRequest, p *Packet, remoteAddr string) (err error)
|
||||
TxRollback(req *proto.TxApplyRequest, p *Packet, remoteAddr string) (err error)
|
||||
TxGetInfo(req *proto.TxGetInfoRequest, p *Packet) (err error)
|
||||
TxGetCnt() (uint64, uint64, uint64)
|
||||
TxGetTree() (*BTree, *BTree, *BTree)
|
||||
@ -194,7 +198,7 @@ type OpExtent interface {
|
||||
BatchObjExtentAppend(req *proto.AppendObjExtentKeysRequest, p *Packet) (err error)
|
||||
ExtentsList(req *proto.GetExtentsRequest, p *Packet) (err error)
|
||||
ObjExtentsList(req *proto.GetExtentsRequest, p *Packet) (err error)
|
||||
ExtentsTruncate(req *ExtentsTruncateReq, p *Packet) (err error)
|
||||
ExtentsTruncate(req *ExtentsTruncateReq, p *Packet, remoteAddr string) (err error)
|
||||
BatchExtentAppend(req *proto.AppendExtentKeysRequest, p *Packet) (err error)
|
||||
// ExtentsDelete(req *proto.DelExtentKeyRequest, p *Packet) (err error)
|
||||
}
|
||||
@ -1557,7 +1561,7 @@ func (mp *metaPartition) delPartitionInodesVersion(verSeq uint64, wg *sync.WaitG
|
||||
Inode: inode.Inode,
|
||||
VerSeq: verSeq,
|
||||
}
|
||||
mp.UnlinkInode(req, p)
|
||||
mp.UnlinkInode(req, p, localAddrForAudit)
|
||||
// check empty result.
|
||||
// if result is OpAgain, means the extDelCh maybe full,
|
||||
// so let it sleep 1s.
|
||||
|
||||
@ -21,11 +21,16 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/cubefs/cubefs/proto"
|
||||
"github.com/cubefs/cubefs/util/auditlog"
|
||||
"github.com/cubefs/cubefs/util/errors"
|
||||
"github.com/cubefs/cubefs/util/log"
|
||||
)
|
||||
|
||||
func (mp *metaPartition) TxCreateDentry(req *proto.TxCreateDentryRequest, p *Packet) (err error) {
|
||||
func (mp *metaPartition) TxCreateDentry(req *proto.TxCreateDentryRequest, p *Packet, remoteAddr string) (err error) {
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogDentryOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.Name, req.FullPath, err, time.Since(start).Milliseconds(), req.Inode, 0)
|
||||
}()
|
||||
if req.ParentID == req.Inode {
|
||||
err = fmt.Errorf("parentId is equal inodeId")
|
||||
p.PacketErrorWithBody(proto.OpExistErr, []byte(err.Error()))
|
||||
@ -76,7 +81,11 @@ func (mp *metaPartition) TxCreateDentry(req *proto.TxCreateDentryRequest, p *Pac
|
||||
}
|
||||
|
||||
// CreateDentry returns a new dentry.
|
||||
func (mp *metaPartition) CreateDentry(req *CreateDentryReq, p *Packet) (err error) {
|
||||
func (mp *metaPartition) CreateDentry(req *CreateDentryReq, p *Packet, remoteAddr string) (err error) {
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogDentryOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.Name, req.FullPath, err, time.Since(start).Milliseconds(), req.Inode, req.ParentID)
|
||||
}()
|
||||
if req.ParentID == req.Inode {
|
||||
err = fmt.Errorf("parentId is equal inodeId")
|
||||
p.PacketErrorWithBody(proto.OpExistErr, []byte(err.Error()))
|
||||
@ -118,7 +127,11 @@ func (mp *metaPartition) CreateDentry(req *CreateDentryReq, p *Packet) (err erro
|
||||
return
|
||||
}
|
||||
|
||||
func (mp *metaPartition) QuotaCreateDentry(req *proto.QuotaCreateDentryRequest, p *Packet) (err error) {
|
||||
func (mp *metaPartition) QuotaCreateDentry(req *proto.QuotaCreateDentryRequest, p *Packet, remoteAddr string) (err error) {
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogDentryOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.Name, req.FullPath, err, time.Since(start).Milliseconds(), req.Inode, req.ParentID)
|
||||
}()
|
||||
if req.ParentID == req.Inode {
|
||||
err = fmt.Errorf("parentId is equal inodeId")
|
||||
p.PacketErrorWithBody(proto.OpExistErr, []byte(err.Error()))
|
||||
@ -169,7 +182,11 @@ func (mp *metaPartition) QuotaCreateDentry(req *proto.QuotaCreateDentryRequest,
|
||||
return
|
||||
}
|
||||
|
||||
func (mp *metaPartition) TxDeleteDentry(req *proto.TxDeleteDentryRequest, p *Packet) (err error) {
|
||||
func (mp *metaPartition) TxDeleteDentry(req *proto.TxDeleteDentryRequest, p *Packet, remoteAddr string) (err error) {
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogDentryOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.Name, req.FullPath, err, time.Since(start).Milliseconds(), req.Ino, req.ParentID)
|
||||
}()
|
||||
txInfo := req.TxInfo.GetCopy()
|
||||
den := &Dentry{
|
||||
ParentId: req.ParentID,
|
||||
@ -240,7 +257,11 @@ func (mp *metaPartition) TxDeleteDentry(req *proto.TxDeleteDentryRequest, p *Pac
|
||||
}
|
||||
|
||||
// DeleteDentry deletes a dentry.
|
||||
func (mp *metaPartition) DeleteDentry(req *DeleteDentryReq, p *Packet) (err error) {
|
||||
func (mp *metaPartition) DeleteDentry(req *DeleteDentryReq, p *Packet, remoteAddr string) (err error) {
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogDentryOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.Name, req.FullPath, err, time.Since(start).Milliseconds(), 0, req.ParentID)
|
||||
}()
|
||||
if req.InodeCreateTime > 0 {
|
||||
if mp.vol.volDeleteLockTime > 0 && req.InodeCreateTime+mp.vol.volDeleteLockTime*60*60 > time.Now().Unix() {
|
||||
err = errors.NewErrorf("the current Inode[%v] is still locked for deletion", req.Name)
|
||||
@ -287,16 +308,24 @@ func (mp *metaPartition) DeleteDentry(req *DeleteDentryReq, p *Packet) (err erro
|
||||
}
|
||||
|
||||
// DeleteDentry deletes a dentry.
|
||||
func (mp *metaPartition) DeleteDentryBatch(req *BatchDeleteDentryReq, p *Packet) (err error) {
|
||||
func (mp *metaPartition) DeleteDentryBatch(req *BatchDeleteDentryReq, p *Packet, remoteAddr string) (err error) {
|
||||
db := make(DentryBatch, 0, len(req.Dens))
|
||||
|
||||
for _, d := range req.Dens {
|
||||
start := time.Now()
|
||||
for i, d := range req.Dens {
|
||||
db = append(db, &Dentry{
|
||||
ParentId: req.ParentID,
|
||||
Name: d.Name,
|
||||
Inode: d.Inode,
|
||||
Type: d.Type,
|
||||
})
|
||||
den := &d
|
||||
fullPath := ""
|
||||
if len(req.FullPaths) > i {
|
||||
fullPath = req.FullPaths[i]
|
||||
}
|
||||
defer func() {
|
||||
auditlog.LogDentryOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), den.Name, fullPath, err, time.Since(start).Milliseconds(), den.Inode, req.ParentID)
|
||||
}()
|
||||
}
|
||||
|
||||
val, err := db.Marshal()
|
||||
@ -349,7 +378,11 @@ func (mp *metaPartition) DeleteDentryBatch(req *BatchDeleteDentryReq, p *Packet)
|
||||
return
|
||||
}
|
||||
|
||||
func (mp *metaPartition) TxUpdateDentry(req *proto.TxUpdateDentryRequest, p *Packet) (err error) {
|
||||
func (mp *metaPartition) TxUpdateDentry(req *proto.TxUpdateDentryRequest, p *Packet, remoteAddr string) (err error) {
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogDentryOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.Name, req.FullPath, err, time.Since(start).Milliseconds(), req.Inode, req.ParentID)
|
||||
}()
|
||||
if req.ParentID == req.Inode {
|
||||
err = fmt.Errorf("parentId is equal inodeId")
|
||||
p.PacketErrorWithBody(proto.OpExistErr, []byte(err.Error()))
|
||||
@ -414,7 +447,11 @@ func (mp *metaPartition) TxUpdateDentry(req *proto.TxUpdateDentryRequest, p *Pac
|
||||
}
|
||||
|
||||
// UpdateDentry updates a dentry.
|
||||
func (mp *metaPartition) UpdateDentry(req *UpdateDentryReq, p *Packet) (err error) {
|
||||
func (mp *metaPartition) UpdateDentry(req *UpdateDentryReq, p *Packet, remoteAddr string) (err error) {
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogDentryOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.Name, req.FullPath, err, time.Since(start).Milliseconds(), req.Inode, req.ParentID)
|
||||
}()
|
||||
if req.ParentID == req.Inode {
|
||||
err = fmt.Errorf("parentId is equal inodeId")
|
||||
p.PacketErrorWithBody(proto.OpExistErr, []byte(err.Error()))
|
||||
|
||||
@ -19,6 +19,10 @@ import (
|
||||
"fmt"
|
||||
"os"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"github.com/cubefs/cubefs/util/auditlog"
|
||||
"github.com/cubefs/cubefs/util/exporter"
|
||||
|
||||
"github.com/cubefs/cubefs/util/exporter"
|
||||
|
||||
@ -454,13 +458,17 @@ func (mp *metaPartition) ObjExtentsList(req *proto.GetExtentsRequest, p *Packet)
|
||||
}
|
||||
|
||||
// ExtentsTruncate truncates an extent.
|
||||
func (mp *metaPartition) ExtentsTruncate(req *ExtentsTruncateReq, p *Packet) (err error) {
|
||||
func (mp *metaPartition) ExtentsTruncate(req *ExtentsTruncateReq, p *Packet, remoteAddr string) (err error) {
|
||||
if !proto.IsHot(mp.volType) {
|
||||
err = fmt.Errorf("only support hot vol")
|
||||
p.PacketErrorWithBody(proto.OpErr, []byte(err.Error()))
|
||||
return
|
||||
}
|
||||
|
||||
fileSize := uint64(0)
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogInodeOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.FullPath, err, time.Since(start).Milliseconds(), req.Inode, fileSize)
|
||||
}()
|
||||
ino := NewInode(req.Inode, proto.Mode(os.ModePerm))
|
||||
item := mp.inodeTree.CopyGet(ino)
|
||||
if item == nil {
|
||||
@ -479,6 +487,7 @@ func (mp *metaPartition) ExtentsTruncate(req *ExtentsTruncateReq, p *Packet) (er
|
||||
}
|
||||
|
||||
ino.Size = req.Size
|
||||
fileSize = ino.Size
|
||||
ino.setVer(mp.verSeq)
|
||||
val, err := ino.Marshal()
|
||||
if err != nil {
|
||||
|
||||
@ -21,6 +21,7 @@ import (
|
||||
"time"
|
||||
|
||||
"github.com/cubefs/cubefs/proto"
|
||||
"github.com/cubefs/cubefs/util/auditlog"
|
||||
"github.com/cubefs/cubefs/util/errors"
|
||||
"github.com/cubefs/cubefs/util/log"
|
||||
)
|
||||
@ -100,14 +101,19 @@ func txReplyInfo(inode *Inode, txInfo *proto.TransactionInfo, quotaInfos map[uin
|
||||
}
|
||||
|
||||
// CreateInode returns a new inode.
|
||||
func (mp *metaPartition) CreateInode(req *CreateInoReq, p *Packet) (err error) {
|
||||
func (mp *metaPartition) CreateInode(req *CreateInoReq, p *Packet, remoteAddr string) (err error) {
|
||||
var (
|
||||
status = proto.OpNotExistErr
|
||||
reply []byte
|
||||
resp interface{}
|
||||
qinode *MetaQuotaInode
|
||||
inoID uint64
|
||||
)
|
||||
inoID, err := mp.nextInodeID()
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogInodeOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.FullPath, err, time.Since(start).Milliseconds(), inoID, 0)
|
||||
}()
|
||||
inoID, err = mp.nextInodeID()
|
||||
if err != nil {
|
||||
p.PacketErrorWithBody(proto.OpInodeFullErr, []byte(err.Error()))
|
||||
return
|
||||
@ -147,14 +153,19 @@ func (mp *metaPartition) CreateInode(req *CreateInoReq, p *Packet) (err error) {
|
||||
return
|
||||
}
|
||||
|
||||
func (mp *metaPartition) QuotaCreateInode(req *proto.QuotaCreateInodeRequest, p *Packet) (err error) {
|
||||
func (mp *metaPartition) QuotaCreateInode(req *proto.QuotaCreateInodeRequest, p *Packet, remoteAddr string) (err error) {
|
||||
var (
|
||||
status = proto.OpNotExistErr
|
||||
reply []byte
|
||||
resp interface{}
|
||||
qinode *MetaQuotaInode
|
||||
inoID uint64
|
||||
)
|
||||
inoID, err := mp.nextInodeID()
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogInodeOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.FullPath, err, time.Since(start).Milliseconds(), inoID, 0)
|
||||
}()
|
||||
inoID, err = mp.nextInodeID()
|
||||
if err != nil {
|
||||
p.PacketErrorWithBody(proto.OpInodeFullErr, []byte(err.Error()))
|
||||
return
|
||||
@ -212,11 +223,14 @@ func (mp *metaPartition) QuotaCreateInode(req *proto.QuotaCreateInodeRequest, p
|
||||
return
|
||||
}
|
||||
|
||||
func (mp *metaPartition) TxUnlinkInode(req *proto.TxUnlinkInodeRequest, p *Packet) (err error) {
|
||||
func (mp *metaPartition) TxUnlinkInode(req *proto.TxUnlinkInodeRequest, p *Packet, remoteAddr string) (err error) {
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogInodeOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.FullPath, err, time.Since(start).Milliseconds(), req.Inode, 0)
|
||||
}()
|
||||
txInfo := req.TxInfo.GetCopy()
|
||||
var status uint8
|
||||
var respIno *Inode
|
||||
|
||||
defer func() {
|
||||
var reply []byte
|
||||
if status == proto.OpOk {
|
||||
@ -293,14 +307,17 @@ func (mp *metaPartition) TxUnlinkInode(req *proto.TxUnlinkInodeRequest, p *Packe
|
||||
}
|
||||
|
||||
// DeleteInode deletes an inode.
|
||||
func (mp *metaPartition) UnlinkInode(req *UnlinkInoReq, p *Packet) (err error) {
|
||||
func (mp *metaPartition) UnlinkInode(req *UnlinkInoReq, p *Packet, remoteAddr string) (err error) {
|
||||
var (
|
||||
msg *InodeResponse
|
||||
reply []byte
|
||||
r interface{}
|
||||
val []byte
|
||||
)
|
||||
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogInodeOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.FullPath, err, time.Since(start).Milliseconds(), req.Inode, 0)
|
||||
}()
|
||||
makeRspFunc := func() {
|
||||
status := msg.Status
|
||||
if status == proto.OpOk {
|
||||
@ -350,16 +367,24 @@ func (mp *metaPartition) UnlinkInode(req *UnlinkInoReq, p *Packet) (err error) {
|
||||
}
|
||||
|
||||
// DeleteInode deletes an inode.
|
||||
func (mp *metaPartition) UnlinkInodeBatch(req *BatchUnlinkInoReq, p *Packet) (err error) {
|
||||
func (mp *metaPartition) UnlinkInodeBatch(req *BatchUnlinkInoReq, p *Packet, remoteAddr string) (err error) {
|
||||
|
||||
if len(req.Inodes) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
var inodes InodeBatch
|
||||
|
||||
for _, id := range req.Inodes {
|
||||
start := time.Now()
|
||||
for i, id := range req.Inodes {
|
||||
inodes = append(inodes, NewInode(id, 0))
|
||||
ino := id
|
||||
fullPath := ""
|
||||
if len(req.FullPaths) > i {
|
||||
fullPath = req.FullPaths[i]
|
||||
}
|
||||
defer func() {
|
||||
auditlog.LogInodeOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), fullPath, err, time.Since(start).Milliseconds(), ino, 0)
|
||||
}()
|
||||
}
|
||||
|
||||
val, err := inodes.Marshal()
|
||||
@ -547,7 +572,11 @@ func (mp *metaPartition) InodeGetBatch(req *InodeGetReqBatch, p *Packet) (err er
|
||||
return
|
||||
}
|
||||
|
||||
func (mp *metaPartition) TxCreateInodeLink(req *proto.TxLinkInodeRequest, p *Packet) (err error) {
|
||||
func (mp *metaPartition) TxCreateInodeLink(req *proto.TxLinkInodeRequest, p *Packet, remoteAddr string) (err error) {
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogInodeOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.FullPath, err, time.Since(start).Milliseconds(), req.Inode, 0)
|
||||
}()
|
||||
txInfo := req.TxInfo.GetCopy()
|
||||
ino := NewInode(req.Inode, 0)
|
||||
inoResp := mp.getInode(ino, true)
|
||||
@ -595,7 +624,11 @@ func (mp *metaPartition) TxCreateInodeLink(req *proto.TxLinkInodeRequest, p *Pac
|
||||
}
|
||||
|
||||
// CreateInodeLink creates an inode link (e.g., soft link).
|
||||
func (mp *metaPartition) CreateInodeLink(req *LinkInodeReq, p *Packet) (err error) {
|
||||
func (mp *metaPartition) CreateInodeLink(req *LinkInodeReq, p *Packet, remoteAddr string) (err error) {
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogInodeOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.FullPath, err, time.Since(start).Milliseconds(), req.Inode, 0)
|
||||
}()
|
||||
var r interface{}
|
||||
var val []byte
|
||||
if req.UniqID > 0 {
|
||||
@ -639,7 +672,11 @@ func (mp *metaPartition) CreateInodeLink(req *LinkInodeReq, p *Packet) (err erro
|
||||
}
|
||||
|
||||
// EvictInode evicts an inode.
|
||||
func (mp *metaPartition) EvictInode(req *EvictInodeReq, p *Packet) (err error) {
|
||||
func (mp *metaPartition) EvictInode(req *EvictInodeReq, p *Packet, remoteAddr string) (err error) {
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogInodeOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.FullPath, err, time.Since(start).Milliseconds(), req.Inode, 0)
|
||||
}()
|
||||
ino := NewInode(req.Inode, 0)
|
||||
val, err := ino.Marshal()
|
||||
if err != nil {
|
||||
@ -657,16 +694,25 @@ func (mp *metaPartition) EvictInode(req *EvictInodeReq, p *Packet) (err error) {
|
||||
}
|
||||
|
||||
// EvictInode evicts an inode.
|
||||
func (mp *metaPartition) EvictInodeBatch(req *BatchEvictInodeReq, p *Packet) (err error) {
|
||||
func (mp *metaPartition) EvictInodeBatch(req *BatchEvictInodeReq, p *Packet, remoteAddr string) (err error) {
|
||||
|
||||
if len(req.Inodes) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
var inodes InodeBatch
|
||||
|
||||
for _, id := range req.Inodes {
|
||||
for i, id := range req.Inodes {
|
||||
inodes = append(inodes, NewInode(id, 0))
|
||||
ino := id
|
||||
fullPath := ""
|
||||
if len(req.FullPaths) > i {
|
||||
fullPath = req.FullPaths[i]
|
||||
}
|
||||
defer func() {
|
||||
auditlog.LogInodeOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), fullPath, err, time.Since(start).Milliseconds(), ino, 0)
|
||||
}()
|
||||
}
|
||||
|
||||
val, err := inodes.Marshal()
|
||||
@ -724,7 +770,11 @@ func (mp *metaPartition) GetInodeTreeLen() int {
|
||||
return mp.inodeTree.Len()
|
||||
}
|
||||
|
||||
func (mp *metaPartition) DeleteInode(req *proto.DeleteInodeRequest, p *Packet) (err error) {
|
||||
func (mp *metaPartition) DeleteInode(req *proto.DeleteInodeRequest, p *Packet, remoteAddr string) (err error) {
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogInodeOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.FullPath, err, time.Since(start).Milliseconds(), req.Inode, 0)
|
||||
}()
|
||||
var bytes = make([]byte, 8)
|
||||
binary.BigEndian.PutUint64(bytes, req.Inode)
|
||||
_, err = mp.submit(opFSMInternalDeleteInode, bytes)
|
||||
@ -736,15 +786,23 @@ func (mp *metaPartition) DeleteInode(req *proto.DeleteInodeRequest, p *Packet) (
|
||||
return
|
||||
}
|
||||
|
||||
func (mp *metaPartition) DeleteInodeBatch(req *proto.DeleteInodeBatchRequest, p *Packet) (err error) {
|
||||
func (mp *metaPartition) DeleteInodeBatch(req *proto.DeleteInodeBatchRequest, p *Packet, remoteAddr string) (err error) {
|
||||
if len(req.Inodes) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
start := time.Now()
|
||||
var inodes InodeBatch
|
||||
|
||||
for _, id := range req.Inodes {
|
||||
for i, id := range req.Inodes {
|
||||
inodes = append(inodes, NewInode(id, 0))
|
||||
ino := id
|
||||
fullPath := ""
|
||||
if len(req.FullPaths) > i {
|
||||
fullPath = req.FullPaths[i]
|
||||
}
|
||||
defer func() {
|
||||
auditlog.LogInodeOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), fullPath, err, time.Since(start).Milliseconds(), ino, 0)
|
||||
}()
|
||||
}
|
||||
|
||||
encoded, err := inodes.Marshal()
|
||||
@ -785,14 +843,18 @@ func (mp *metaPartition) ClearInodeCache(req *proto.ClearInodeCacheRequest, p *P
|
||||
}
|
||||
|
||||
// TxCreateInode returns a new inode.
|
||||
func (mp *metaPartition) TxCreateInode(req *proto.TxCreateInodeRequest, p *Packet) (err error) {
|
||||
func (mp *metaPartition) TxCreateInode(req *proto.TxCreateInodeRequest, p *Packet, remoteAddr string) (err error) {
|
||||
var (
|
||||
status = proto.OpNotExistErr
|
||||
reply []byte
|
||||
resp interface{}
|
||||
inoID uint64
|
||||
)
|
||||
|
||||
inoID, err := mp.nextInodeID()
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogInodeOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.FullPath, err, time.Since(start).Milliseconds(), inoID, 0)
|
||||
}()
|
||||
inoID, err = mp.nextInodeID()
|
||||
if err != nil {
|
||||
p.PacketErrorWithBody(proto.OpInodeFullErr, []byte(err.Error()))
|
||||
return
|
||||
|
||||
@ -18,8 +18,10 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/cubefs/cubefs/proto"
|
||||
"github.com/cubefs/cubefs/util/auditlog"
|
||||
"github.com/cubefs/cubefs/util/log"
|
||||
)
|
||||
|
||||
@ -224,7 +226,12 @@ func (mp *metaPartition) TxRollbackRM(req *proto.TxApplyRMRequest, p *Packet) er
|
||||
return nil
|
||||
}
|
||||
|
||||
func (mp *metaPartition) TxCommit(req *proto.TxApplyRequest, p *Packet) error {
|
||||
func (mp *metaPartition) TxCommit(req *proto.TxApplyRequest, p *Packet, remoteAddr string) error {
|
||||
var err error
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogTxOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.TxID, err, time.Since(start).Milliseconds())
|
||||
}()
|
||||
status, err := mp.txProcessor.txManager.commitTx(req.TxID, false)
|
||||
if err != nil {
|
||||
p.PacketErrorWithBody(status, []byte(err.Error()))
|
||||
@ -234,8 +241,12 @@ func (mp *metaPartition) TxCommit(req *proto.TxApplyRequest, p *Packet) error {
|
||||
return err
|
||||
}
|
||||
|
||||
func (mp *metaPartition) TxRollback(req *proto.TxApplyRequest, p *Packet) error {
|
||||
|
||||
func (mp *metaPartition) TxRollback(req *proto.TxApplyRequest, p *Packet, remoteAddr string) error {
|
||||
var err error
|
||||
start := time.Now()
|
||||
defer func() {
|
||||
auditlog.LogTxOp(remoteAddr, mp.GetVolName(), p.GetOpMsg(), req.TxID, err, time.Since(start).Milliseconds())
|
||||
}()
|
||||
status, err := mp.txProcessor.txManager.rollbackTx(req.TxID, false)
|
||||
if err != nil {
|
||||
p.PacketErrorWithBody(status, []byte(err.Error()))
|
||||
|
||||
@ -192,6 +192,7 @@ type QuotaCreateInodeRequest struct {
|
||||
Gid uint32 `json:"gid"`
|
||||
Target []byte `json:"tgt"`
|
||||
QuotaIds []uint32 `json:"qids"`
|
||||
FullPath string `json:"fullPath"`
|
||||
}
|
||||
|
||||
type CreateInodeRequest struct {
|
||||
@ -201,6 +202,7 @@ type CreateInodeRequest struct {
|
||||
Uid uint32 `json:"uid"`
|
||||
Gid uint32 `json:"gid"`
|
||||
Target []byte `json:"tgt"`
|
||||
FullPath string `json:"fullPath"`
|
||||
}
|
||||
|
||||
// CreateInodeResponse defines the response to the request of creating an inode.
|
||||
@ -234,6 +236,7 @@ type TxCreateInodeRequest struct {
|
||||
Target []byte `json:"tgt"`
|
||||
QuotaIds []uint32 `json:"qids"`
|
||||
TxInfo *TransactionInfo `json:"tx"`
|
||||
FullPath string `json:"fullPath"`
|
||||
}
|
||||
|
||||
// TxCreateInodeResponse defines the response with transaction info to the request of creating an inode.
|
||||
@ -291,6 +294,7 @@ type LinkInodeRequest struct {
|
||||
Inode uint64 `json:"ino"`
|
||||
UniqID uint64 `json:"uiq"`
|
||||
IsRename bool `json:"rename"`
|
||||
FullPath string `json:"fullPath"`
|
||||
}
|
||||
|
||||
// LinkInodeResponse defines the response to the request of linking an inode.
|
||||
@ -303,6 +307,7 @@ type TxLinkInodeRequest struct {
|
||||
PartitionID uint64 `json:"pid"`
|
||||
Inode uint64 `json:"ino"`
|
||||
TxInfo *TransactionInfo `json:"tx"`
|
||||
FullPath string `json:"fullPath"`
|
||||
}
|
||||
|
||||
func (tx *TxLinkInodeRequest) GetInfo() string {
|
||||
@ -329,6 +334,7 @@ type TxUnlinkInodeRequest struct {
|
||||
Inode uint64 `json:"ino"`
|
||||
Evict bool `json:"evict"`
|
||||
TxInfo *TransactionInfo `json:"tx"`
|
||||
FullPath string `json:"fullPath"`
|
||||
}
|
||||
|
||||
func (tx *TxUnlinkInodeRequest) GetInfo() string {
|
||||
@ -348,6 +354,7 @@ type UnlinkInodeRequest struct {
|
||||
UniqID uint64 `json:"uid"` //for request dedup
|
||||
VerSeq uint64 `json:"ver"`
|
||||
DenVerSeq uint64 `json:"denVer"`
|
||||
FullPath string `json:"fullPath"`
|
||||
}
|
||||
|
||||
// UnlinkInodeRequest defines the request to unlink an inode.
|
||||
@ -355,6 +362,7 @@ type BatchUnlinkInodeRequest struct {
|
||||
VolName string `json:"vol"`
|
||||
PartitionID uint64 `json:"pid"`
|
||||
Inodes []uint64 `json:"inos"`
|
||||
FullPaths []string `json:"fullPaths"`
|
||||
}
|
||||
|
||||
// UnlinkInodeResponse defines the response to the request of unlinking an inode.
|
||||
@ -375,6 +383,7 @@ type EvictInodeRequest struct {
|
||||
VolName string `json:"vol"`
|
||||
PartitionID uint64 `json:"pid"`
|
||||
Inode uint64 `json:"ino"`
|
||||
FullPath string `json:"fullPath"`
|
||||
}
|
||||
|
||||
// EvictInodeRequest defines the request to evict some inode.
|
||||
@ -382,6 +391,7 @@ type BatchEvictInodeRequest struct {
|
||||
VolName string `json:"vol"`
|
||||
PartitionID uint64 `json:"pid"`
|
||||
Inodes []uint64 `json:"inos"`
|
||||
FullPaths []string `json:"fullPaths"`
|
||||
}
|
||||
|
||||
// CreateDentryRequest defines the request to create a dentry.
|
||||
@ -394,6 +404,7 @@ type QuotaCreateDentryRequest struct {
|
||||
Mode uint32 `json:"mode"`
|
||||
QuotaIds []uint32 `json:"qids"`
|
||||
VerSeq uint64 `json:"seq"`
|
||||
FullPath string `json:"fullPath"`
|
||||
}
|
||||
|
||||
type CreateDentryRequest struct {
|
||||
@ -404,6 +415,7 @@ type CreateDentryRequest struct {
|
||||
Name string `json:"name"`
|
||||
Mode uint32 `json:"mode"`
|
||||
VerSeq uint64 `json:"seq"`
|
||||
FullPath string `json:"fullPath"`
|
||||
}
|
||||
|
||||
type TxPack interface {
|
||||
@ -420,6 +432,7 @@ type TxCreateDentryRequest struct {
|
||||
Mode uint32 `json:"mode"`
|
||||
QuotaIds []uint32 `json:"qids"`
|
||||
TxInfo *TransactionInfo `json:"tx"`
|
||||
FullPath string `json:"fullPath"`
|
||||
}
|
||||
|
||||
func (tx *TxCreateDentryRequest) GetInfo() string {
|
||||
@ -437,6 +450,7 @@ type UpdateDentryRequest struct {
|
||||
ParentID uint64 `json:"pino"`
|
||||
Name string `json:"name"`
|
||||
Inode uint64 `json:"ino"` // new inode number
|
||||
FullPath string `json:"fullPath"`
|
||||
}
|
||||
|
||||
// UpdateDentryResponse defines the response to the request of updating a dentry.
|
||||
@ -452,6 +466,7 @@ type TxUpdateDentryRequest struct {
|
||||
Inode uint64 `json:"ino"` // new inode number
|
||||
OldIno uint64 `json:"oldIno"` // new inode number
|
||||
TxInfo *TransactionInfo `json:"tx"`
|
||||
FullPath string `json:"fullPath"`
|
||||
}
|
||||
|
||||
func (tx *TxUpdateDentryRequest) GetInfo() string {
|
||||
@ -469,6 +484,7 @@ type TxDeleteDentryRequest struct {
|
||||
Name string `json:"name"`
|
||||
Ino uint64 `json:"ino"`
|
||||
TxInfo *TransactionInfo `json:"tx"`
|
||||
FullPath string `json:"fullPath"`
|
||||
}
|
||||
|
||||
func (tx *TxDeleteDentryRequest) GetInfo() string {
|
||||
@ -487,6 +503,7 @@ type DeleteDentryRequest struct {
|
||||
Name string `json:"name"`
|
||||
InodeCreateTime int64 `json:"inodeCreateTime"`
|
||||
Verseq uint64 `json:"ver"`
|
||||
FullPath string `json:"fullPath"`
|
||||
}
|
||||
|
||||
type BatchDeleteDentryRequest struct {
|
||||
@ -494,6 +511,7 @@ type BatchDeleteDentryRequest struct {
|
||||
PartitionID uint64 `json:"pid"`
|
||||
ParentID uint64 `json:"pino"`
|
||||
Dens []Dentry `json:"dens"`
|
||||
FullPaths []string `json:"fullPaths"`
|
||||
}
|
||||
|
||||
// DeleteDentryResponse defines the response to the request of deleting a dentry.
|
||||
@ -679,6 +697,7 @@ type TruncateRequest struct {
|
||||
PartitionID uint64 `json:"pid"`
|
||||
Inode uint64 `json:"ino"`
|
||||
Size uint64 `json:"sz"`
|
||||
FullPath string `json:"fullPath"`
|
||||
}
|
||||
|
||||
type EmptyExtentKeyRequest struct {
|
||||
@ -728,6 +747,7 @@ type DeleteInodeRequest struct {
|
||||
VolName string `json:"vol"`
|
||||
PartitionId uint64 `json:"pid"`
|
||||
Inode uint64 `json:"ino"`
|
||||
FullPath string `json:"fullPath"`
|
||||
}
|
||||
|
||||
// DeleteInodeRequest defines the request to delete an inode.
|
||||
@ -735,6 +755,7 @@ type DeleteInodeBatchRequest struct {
|
||||
VolName string `json:"vol"`
|
||||
PartitionId uint64 `json:"pid"`
|
||||
Inodes []uint64 `json:"ino"`
|
||||
FullPaths []string `json:"fullPaths"`
|
||||
}
|
||||
|
||||
// AppendExtentKeysRequest defines the request to append an extent key.
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
