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

audit log

Browse Source 
Signed-off-by: Patrick Wu <wuchunhuan@oppo.com>
This commit is contained in:
Patrick Wu 2022-09-14 14:34:33 +08:00 committed by leonrayang
parent fe9c35bb37
commit de36b418cf
8 changed files with 625 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
client/fs/dir.go
View File

@ -16,6 +16,7 @@ package fs
import (
"fmt"
"github.com/cubefs/cubefs/util/auditlog"
"io"
"os"
"strconv"
@ -155,10 +156,12 @@ func (d *Dir) Create(ctx context.Context, req *fuse.CreateRequest, resp *fuse.Cr
bgTime := stat.BeginStat()
var err error
var newInode uint64
metric := exporter.NewTPCnt("filecreate")
defer func() {
stat.EndStat("Create", err, bgTime, 1)
metric.SetWithLabels(err, map[string]string{exporter.Vol: d.super.volname})
auditlog.FormatLog("Create", d.getCwd()+"/"+req.Name, "nil", err, time.Since(start).Microseconds(), newInode, 0)
}()
info, err := d.super.mw.Create_ll(d.info.Inode, req.Name, proto.Mode(req.Mode.Perm()), req.Uid, req.Gid, nil)
@ -169,6 +172,7 @@ func (d *Dir) Create(ctx context.Context, req *fuse.CreateRequest, resp *fuse.Cr
d.super.ic.Put(info)
child := NewFile(d.super, info, uint32(req.Flags&DefaultFlag), d.info.Inode, req.Name)
newInode = info.Inode
d.super.ec.OpenStream(info.Inode)
d.super.fslock.Lock()
@ -209,10 +213,12 @@ func (d *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, error
bgTime := stat.BeginStat()
var err error
var newInode uint64
metric := exporter.NewTPCnt("mkdir")
defer func() {
stat.EndStat("Mkdir", err, bgTime, 1)
metric.SetWithLabels(err, map[string]string{exporter.Vol: d.super.volname})
auditlog.FormatLog("Mkdir", d.getCwd()+"/"+req.Name, "nil", err, time.Since(start).Microseconds(), newInode, 0)
}()
info, err := d.super.mw.Create_ll(d.info.Inode, req.Name, proto.Mode(os.ModeDir|req.Mode.Perm()), req.Uid, req.Gid, nil)
@ -223,7 +229,7 @@ func (d *Dir) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Node, error
d.super.ic.Put(info)
child := NewDir(d.super, info, d.info.Inode, req.Name)
newInode = info.Inode
d.super.fslock.Lock()
d.super.nodeCache[info.Inode] = child
d.super.fslock.Unlock()
@ -244,10 +250,12 @@ func (d *Dir) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
bgTime := stat.BeginStat()
var err error
var deletedInode uint64
metric := exporter.NewTPCnt("remove")
defer func() {
stat.EndStat("Remove", err, bgTime, 1)
metric.SetWithLabels(err, map[string]string{exporter.Vol: d.super.volname})
auditlog.FormatLog("Remove", d.getCwd()+"/"+req.Name, "nil", err, time.Since(start).Microseconds(), deletedInode, 0)
}()
info, err := d.super.mw.Delete_ll(d.info.Inode, req.Name, req.Dir)
@ -256,6 +264,9 @@ func (d *Dir) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
return ParseError(err)
}
if info != nil {
deletedInode = info.Inode
}
d.super.ic.Delete(d.info.Inode)
if info != nil && info.Nlink == 0 && !proto.IsDir(info.Mode) {
@ -537,6 +548,7 @@ func (d *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDir fs.Nod
defer func() {
stat.EndStat("Rename", err, bgTime, 1)
metric.SetWithLabels(err, map[string]string{exporter.Vol: d.super.volname})
auditlog.FormatLog("Rename", d.getCwd()+"/"+req.OldName, dstDir.getCwd()+"/"+req.NewName, err, time.Since(start).Microseconds(), 0, 0)
}()
err = d.super.mw.Rename_ll(d.info.Inode, req.OldName, dstDir.info.Inode, req.NewName, true)

6
client/fs/file.go
View File

@ -65,7 +65,7 @@ var (
)
// NewFile returns a new file.
func NewFile(s *Super, i *proto.InodeInfo, flag uint32, pino uint64, name string) fs.Node {
func NewFile(s *Super, i *proto.InodeInfo, flag uint32, pino uint64, filename string) fs.Node {
if proto.IsCold(s.volType) {
var (
fReader *blobstore.Reader
@ -103,9 +103,9 @@ func NewFile(s *Super, i *proto.InodeInfo, flag uint32, pino uint64, name string
fWriter = blobstore.NewWriter(clientConf)
}
log.LogDebugf("Trace NewFile:fReader(%v) fWriter(%v) ", fReader, fWriter)
return &File{super: s, info: i, fWriter: fWriter, fReader: fReader, parentIno: pino}
return &File{super: s, info: i, fWriter: fWriter, fReader: fReader, parentIno: pino, name: filename}
}
return &File{super: s, info: i, parentIno: pino, name: name}
return &File{super: s, info: i, parentIno: pino, name: filename}
}
//get file parentPath

24
client/fuse.go
View File

@ -23,6 +23,8 @@ package main
import (
"flag"
"fmt"
"github.com/cubefs/cubefs/blockcache/bcache"
"github.com/cubefs/cubefs/util/auditlog"
"io/ioutil"
syslog "log"
"net"
@ -38,8 +40,6 @@ import (
"syscall"
"time"
"github.com/cubefs/cubefs/blockcache/bcache"
"github.com/cubefs/cubefs/util/buf"
"github.com/cubefs/cubefs/sdk/master"
@ -321,6 +321,16 @@ func main() {
}
stat.ClearStat()
if opt.EnableAudit {
_, err = auditlog.InitAudit(opt.Logpath, LoggerPrefix, int64(auditlog.DefaultAuditLogSize))
if err != nil {
err = errors.NewErrorf("Init audit log fail: %v\n", err)
fmt.Println(err)
daemonize.SignalOutcome(err)
os.Exit(1)
}
}
proto.InitBufferPool(opt.BuffersTotalLimit)
if proto.IsCold(opt.VolType) {
buf.InitCachePool(opt.EbsBlockSize)
@ -407,6 +417,10 @@ func main() {
syslog.Printf("enable bcache %v", opt.EnableBcache)
if cfg.GetString(exporter.ConfigKeyPushAddr) == "" {
cfg.SetString(exporter.ConfigKeyPushAddr, "cfs-push.oppo.local")
}
exporter.Init(ModuleName, cfg)
exporter.RegistConsul(super.ClusterName(), ModuleName, cfg)
@ -560,6 +574,10 @@ func mount(opt *proto.MountOptions) (fsConn *fuse.Conn, super *cfs.Super, err er
http.HandleFunc(log.GetLogPath, log.GetLog)
http.HandleFunc(ControlCommandSuspend, super.SetSuspend)
http.HandleFunc(ControlCommandResume, super.SetResume)
//auditlog
http.HandleFunc(auditlog.EnableAuditLogReqPath, auditlog.EnableAuditLog)
http.HandleFunc(auditlog.DisableAuditLogReqPath, auditlog.DisableAuditLog)
http.HandleFunc(auditlog.SetAuditLogBufSizeReqPath, auditlog.ResetWriterBuffSize)
statusCh := make(chan error)
go waitListenAndServe(statusCh, ":"+opt.Profport, nil)
@ -632,6 +650,7 @@ func registerInterceptedSignal(mnt string) {
go func() {
sig := <-sigC
syslog.Printf("Killed due to a received signal (%v)\n", sig)
auditlog.StopAudit()
os.Exit(1)
}()
}
@ -706,6 +725,7 @@ func parseMountOption(cfg *config.Config) (*proto.MountOptions, error) {
opt.BuffersTotalLimit = GlobalMountOptions[proto.BuffersTotalLimit].GetInt64()
opt.MetaSendTimeout = GlobalMountOptions[proto.MetaSendTimeout].GetInt64()
opt.MaxStreamerLimit = GlobalMountOptions[proto.MaxStreamerLimit].GetInt64()
opt.EnableAudit = GlobalMountOptions[proto.EnableAudit].GetBool()
if opt.MountPoint == "" || opt.Volname == "" || opt.Owner == "" || opt.Master == "" {
return nil, errors.New(fmt.Sprintf("invalid config file: lack of mandatory fields, mountPoint(%v), volName(%v), owner(%v), masterAddr(%v)", opt.MountPoint, opt.Volname, opt.Owner, opt.Master))

3
docker/conf/client.json
View File

@ -12,5 +12,6 @@
"ticketHost": "192.168.0.14:8080,192.168.0.15:8081,192.168.0.16:8082",
"enableHTTPS": "false",
"accessKey": "39bEF4RrAQgMj6RV",
"secretKey": "TRL6o3JL16YOqvZGIohBDFTHZDEcFsyd"
"secretKey": "TRL6o3JL16YOqvZGIohBDFTHZDEcFsyd",
"enableAudit": true
}

1
docker/docker-compose.yml
View File

@ -347,6 +347,7 @@ services:
image: ghcr.io/cubefs/cbfs-base:1.0-golang-1.16.12
ports:
- 9500
- 17410:17410
volumes:
- ./bin:/cfs/bin:ro
- ./conf/hosts:/etc/hosts:ro

75
libsdk/libsdk.go
View File

@ -76,6 +76,7 @@ import "C"
import (
"context"
"fmt"
"github.com/cubefs/cubefs/util/auditlog"
"github.com/cubefs/cubefs/util/buf"
"io"
syslog "log"
@ -88,6 +89,7 @@ import (
"sync"
"sync/atomic"
"syscall"
"time"
"unsafe"
"github.com/cubefs/blobstore/api/access"
@ -238,6 +240,8 @@ type client struct {
pushAddr string
cluster string
dirChildrenNumLimit uint32
enableAudit bool
// runtime context
cwd string // current working directory
fdmap map[uint]*file
@ -314,6 +318,12 @@ func cfs_set_client(id C.int64_t, key, val *C.char) C.int {
c.secretKey = v
case "pushAddr":
c.pushAddr = v
case "enableAudit":
if v == "true" {
c.enableAudit = true
} else {
c.enableAudit = false
}
default:
return statusEINVAL
}
@ -346,6 +356,7 @@ func cfs_close_client(id C.int64_t) {
}
removeClient(int64(id))
}
auditlog.StopAudit()
log.LogFlush()
}
@ -455,6 +466,7 @@ func cfs_open(id C.int64_t, path *C.char, flags C.int, mode C.mode_t) C.int {
if !exist {
return statusEINVAL
}
start := time.Now()
fuseMode := uint32(mode) & uint32(0777)
fuseFlags := uint32(flags) &^ uint32(0x8000)
@ -479,6 +491,13 @@ func cfs_open(id C.int64_t, path *C.char, flags C.int, mode C.mode_t) C.int {
return errorToStatus(err)
}
parentIno = dirInfo.Inode
defer func() {
if info != nil {
auditlog.FormatLog("Create", dirpath, "nil", err, time.Since(start).Microseconds(), info.Inode, 0)
} else {
auditlog.FormatLog("Create", dirpath, "nil", err, time.Since(start).Microseconds(), 0, 0)
}
}()
newInfo, err := c.create(dirInfo.Inode, name, fuseMode)
if err != nil {
if err != syscall.EEXIST {
@ -889,11 +908,23 @@ func cfs_mkdirs(id C.int64_t, path *C.char, mode C.mode_t) C.int {
return statusEINVAL
}
start := time.Now()
var gerr error
var gino uint64
dirpath := c.absPath(C.GoString(path))
if dirpath == "/" {
return statusEEXIST
}
defer func() {
if gerr == nil {
auditlog.FormatLog("Mkdir", dirpath, "nil", gerr, time.Since(start).Microseconds(), gino, 0)
} else {
auditlog.FormatLog("Mkdir", dirpath, "nil", gerr, time.Since(start).Microseconds(), 0, 0)
}
}()
pino := proto.RootIno
dirs := strings.Split(dirpath, "/")
for _, dir := range dirs {
@ -907,16 +938,19 @@ func cfs_mkdirs(id C.int64_t, path *C.char, mode C.mode_t) C.int {
if err != nil {
if err != syscall.EEXIST {
gerr = err
return errorToStatus(err)
}
} else {
child = info.Inode
}
} else {
gerr = err
return errorToStatus(err)
}
}
pino = child
gino = child
}
return 0
@ -928,15 +962,25 @@ func cfs_rmdir(id C.int64_t, path *C.char) C.int {
if !exist {
return statusEINVAL
}
start := time.Now()
var err error
var info *proto.InodeInfo
absPath := c.absPath(C.GoString(path))
defer func() {
if info == nil {
auditlog.FormatLog("Rmdir", absPath, "nil", err, time.Since(start).Microseconds(), 0, 0)
} else {
auditlog.FormatLog("Rmdir", absPath, "nil", err, time.Since(start).Microseconds(), info.Inode, 0)
}
}()
dirpath, name := gopath.Split(absPath)
dirInfo, err := c.lookupPath(dirpath)
if err != nil {
return errorToStatus(err)
}
_, err = c.mw.Delete_ll(dirInfo.Inode, name, true)
info, err = c.mw.Delete_ll(dirInfo.Inode, name, true)
c.ic.Delete(dirInfo.Inode)
c.dc.Delete(absPath)
return errorToStatus(err)
@ -949,8 +993,20 @@ func cfs_unlink(id C.int64_t, path *C.char) C.int {
return statusEINVAL
}
start := time.Now()
var err error
var info *proto.InodeInfo
absPath := c.absPath(C.GoString(path))
dirpath, name := gopath.Split(absPath)
defer func() {
if info == nil {
auditlog.FormatLog("Unlink", absPath, "nil", err, time.Since(start).Microseconds(), 0, 0)
} else {
auditlog.FormatLog("Unlink", absPath, "nil", err, time.Since(start).Microseconds(), info.Inode, 0)
}
}()
dirInfo, err := c.lookupPath(dirpath)
if err != nil {
return errorToStatus(err)
@ -964,7 +1020,7 @@ func cfs_unlink(id C.int64_t, path *C.char) C.int {
return statusEISDIR
}
info, err := c.mw.Delete_ll(dirInfo.Inode, name, false)
info, err = c.mw.Delete_ll(dirInfo.Inode, name, false)
if err != nil {
return errorToStatus(err)
}
@ -983,8 +1039,16 @@ func cfs_rename(id C.int64_t, from *C.char, to *C.char) C.int {
return statusEINVAL
}
start := time.Now()
var err error
absFrom := c.absPath(C.GoString(from))
absTo := c.absPath(C.GoString(to))
defer func() {
auditlog.FormatLog("Rename", absFrom, absTo, err, time.Since(start).Microseconds(), 0, 0)
}()
srcDirPath, srcName := gopath.Split(absFrom)
dstDirPath, dstName := gopath.Split(absTo)
@ -1104,6 +1168,13 @@ func (c *client) start() (err error) {
return
}
if c.enableAudit {
_, err = auditlog.InitAudit(c.logDir, "clientSdk", int64(auditlog.DefaultAuditLogSize))
if err != nil {
log.LogWarnf("Init audit log fail: %v", err)
}
}
if c.enableSummary {
c.sc = fs.NewSummaryCache(fs.DefaultSummaryExpiration, fs.MaxSummaryCache)
}

4
proto/mount_options.go
View File

@ -66,6 +66,8 @@ const (
BuffersTotalLimit
MaxStreamerLimit
EnableAudit
MaxMountOption
)
@ -149,6 +151,7 @@ func InitMountOptions(opts []MountOption) {
opts[BcacheFilterFiles] = MountOption{"bcacheFilterFiles", "The block cache filter files suffix", "", "py;pyx;sh;yaml;conf"}
opts[BcacheBatchCnt] = MountOption{"bcacheBatchCnt", "The block cache get meta count", "", int64(100000)}
opts[BcacheCheckIntervalS] = MountOption{"bcacheCheckIntervalS", "The block cache check interval", "", int64(300)}
opts[EnableAudit] = MountOption{"enableAudit", "enable client audit logging", "", false}
for i := 0; i < MaxMountOption; i++ {
flag.StringVar(&opts[i].cmdlineValue, opts[i].keyword, "", opts[i].description)
@ -297,4 +300,5 @@ type MountOptions struct {
MetaSendTimeout int64
BuffersTotalLimit int64
MaxStreamerLimit int64
EnableAudit bool
}

507
util/auditlog/auditlog.go Normal file
View File

@ -0,0 +1,507 @@
package auditlog
import "C"
import (
"bufio"
"encoding/json"
"errors"
"fmt"
"github.com/cubefs/cubefs/util/log"
"io/ioutil"
"net"
"net/http"
"os"
"path"
"regexp"
"sort"
"strconv"
"strings"
"sync"
"syscall"
"time"
)
const (
Audit_Module = "audit"
FileNameDateFormat = "20060102150405"
ShiftedExtension = ".old"
DefaultAuditLogBufSize = 4096
F_OK = 0
DefaultCleanInterval = 1 * time.Hour
DefaultAuditLogSize = 200 * 1024 * 1024 // 200M
DefaultHeadRoom = 50 * 1024 // 50G
MaxReservedDays = 7 * 24 * time.Hour
)
const (
EnableAuditLogReqPath = "/auditlog/enable"
DisableAuditLogReqPath = "/auditlog/disable"
SetAuditLogBufSizeReqPath = "/auditlog/setbufsize"
)
var DefaultTimeOutUs = [3]uint32{100000, 500000, 1000000}
var re = regexp.MustCompile(`\([0-9]*\)`)
type ShiftedFile []os.FileInfo
func (f ShiftedFile) Less(i, j int) bool {
return f[i].ModTime().Before(f[j].ModTime())
}
func (f ShiftedFile) Len() int {
return len(f)
}
func (f ShiftedFile) Swap(i, j int) {
f[i], f[j] = f[j], f[i]
}
//type typeInfo struct {
// typeName string
// allCount uint32
// failCount uint32
// maxTime time.Duration
// minTime time.Duration
// allTimeUs time.Duration
// timeOut [MaxTimeoutLevel]uint32
//}
type Audit struct {
volName string
hostName string
ipAddr string
logDir string
logModule string
logMaxSize int64
logFileName string
logFile *os.File
writer *bufio.Writer
writerBufSize int
bufferC chan string
stopC chan struct{}
resetWriterBuffC chan int
pid int
//lastClearTime time.Time
//timeOutUs [MaxTimeoutLevel]uint32
//typeInfoMap map[string]*typeInfo
//closeAudit bool
//useMutex bool
//sync.Mutex
}
var gAdt *Audit = nil
var AdtMutex sync.RWMutex
var once sync.Once
func getAddr() (HostName, IPAddr string) {
hostName, err := os.Hostname()
if err != nil {
HostName = "Unknown"
log.LogWarnf("Get host name failed, replaced by unknown. err(%v)", err)
} else {
HostName = hostName
}
addrs, err := net.InterfaceAddrs()
if err != nil {
IPAddr = "Unknown"
log.LogWarnf("Get ip address failed, replaced by unknown. err(%v)", err)
} else {
var ip_addrs []string
for _, addr := range addrs {
if ipnet, ok := addr.(*net.IPNet); ok && !ipnet.IP.IsLoopback() && ipnet.IP.To4() != nil {
ip_addrs = append(ip_addrs, ipnet.IP.String())
}
}
if len(ip_addrs) > 0 {
IPAddr = strings.Join(ip_addrs, ",")
} else {
IPAddr = "Unknown"
log.LogWarnf("Get ip address failed, replaced by unknown. err(%v)", err)
}
}
return
}
func ResetWriterBuffSize(w http.ResponseWriter, r *http.Request) {
var (
err error
)
if err = r.ParseForm(); err != nil {
buildFailureResp(w, http.StatusBadRequest, err.Error())
return
}
size := int(DefaultAuditLogBufSize)
if sizeStr := r.FormValue("size"); sizeStr != "" {
val, err := strconv.Atoi(sizeStr)
if err != nil {
err = fmt.Errorf("size error")
buildFailureResp(w, http.StatusBadRequest, err.Error())
return
}
size = val
}
ResetWriterBufferSize(size)
buildSuccessResp(w, "set audit log buffer size success")
}
func DisableAuditLog(w http.ResponseWriter, r *http.Request) {
StopAudit()
buildSuccessResp(w, "disable audit log success")
}
func EnableAuditLog(w http.ResponseWriter, r *http.Request) {
var (
err error
)
if err = r.ParseForm(); err != nil {
buildFailureResp(w, http.StatusBadRequest, err.Error())
return
}
logPath := r.FormValue("path")
if logPath == "" {
err = fmt.Errorf("path cannot be empty")
buildFailureResp(w, http.StatusBadRequest, err.Error())
return
}
prefix := r.FormValue("prefix")
if prefix == "" {
err = fmt.Errorf("prefix cannot be empty")
buildFailureResp(w, http.StatusBadRequest, err.Error())
return
}
logSize := 0
if logSizeStr := r.FormValue("logsize"); logSizeStr != "" {
val, err := strconv.Atoi(logSizeStr)
if err != nil {
err = fmt.Errorf("logSize error")
buildFailureResp(w, http.StatusBadRequest, err.Error())
return
}
logSize = val
} else {
logSize = DefaultAuditLogSize
}
err, dir, logModule, logMaxSize := GetAuditLogInfo()
if err != nil {
_, err = InitAudit(logPath, prefix, int64(logSize))
if err != nil {
err = fmt.Errorf("Init audit log fail: %v\n", err)
buildFailureResp(w, http.StatusBadRequest, err.Error())
return
}
info := fmt.Sprintf("audit log is initialized with params: logDir(%v) logModule(%v) logMaxSize(%v)",
logPath, prefix, logSize)
buildSuccessResp(w, info)
} else {
info := fmt.Sprintf("audit log is already initialized with params: logDir(%v) logModule(%v) logMaxSize(%v)",
dir, logModule, logMaxSize)
buildSuccessResp(w, info)
}
}
func buildSuccessResp(w http.ResponseWriter, data interface{}) {
buildJSONResp(w, http.StatusOK, data, "")
}
func buildFailureResp(w http.ResponseWriter, code int, msg string) {
buildJSONResp(w, code, nil, msg)
}
// Create response for the API request.
func buildJSONResp(w http.ResponseWriter, code int, data interface{}, msg string) {
var (
jsonBody []byte
err error
)
w.WriteHeader(code)
w.Header().Set("Content-Type", "application/json")
body := struct {
Code int `json:"code"`
Data interface{} `json:"data"`
Msg string `json:"msg"`
}{
Code: code,
Data: data,
Msg: msg,
}
if jsonBody, err = json.Marshal(body); err != nil {
return
}
w.Write(jsonBody)
}
func GetAuditLogInfo() (err error, dir, logModule string, logMaxSize int64) {
AdtMutex.RLock()
defer AdtMutex.RUnlock()
if gAdt != nil {
return nil, gAdt.logDir, gAdt.logModule, gAdt.logMaxSize
} else {
return errors.New("audit log is not initialized yet"), "", "", 0
}
}
func InitAudit(dir, logModule string, logMaxSize int64) (*Audit, error) {
AdtMutex.Lock()
defer AdtMutex.Unlock()
if gAdt != nil {
return gAdt, nil
}
if dir == "" || logModule == "" || logMaxSize <= 0 {
errInfo := fmt.Sprintf("InitAudit failed with params: dir(%v) logModule(%v) logMaxSize(%v)",
dir, logModule, logMaxSize)
return nil, errors.New(errInfo)
}
host, ip := getAddr()
dir = path.Join(dir, logModule)
fi, err := os.Stat(dir)
if err != nil {
os.MkdirAll(dir, 0755)
} else {
if !fi.IsDir() {
return nil, errors.New(dir + " is not a directory")
}
}
_ = os.Chmod(dir, 0766)
logName := path.Join(dir, Audit_Module) + ".log"
audit := &Audit{
hostName: host,
ipAddr: ip,
logDir: dir,
logModule: logModule,
logMaxSize: logMaxSize,
logFileName: logName,
writerBufSize: DefaultAuditLogBufSize,
bufferC: make(chan string, 1000),
stopC: make(chan struct{}),
resetWriterBuffC: make(chan int),
pid: os.Getpid(),
}
err = audit.newWriterSize(audit.writerBufSize)
if err != nil {
return nil, err
}
gAdt = audit
go audit.flushAuditLog()
return audit, nil
}
func formatAuditEntry(op, src, dst string, err error, latency int64, srcInode, dstInode uint64) (entry string) {
AdtMutex.RLock()
if gAdt == nil {
AdtMutex.RUnlock()
return ""
}
ipAddr := gAdt.ipAddr
hostName := gAdt.hostName
AdtMutex.RUnlock()
var errStr string
if err != nil {
errStr = err.Error()
} else {
errStr = "nil"
}
curTime := time.Now()
curTimeStr := curTime.Format("2006-01-02 15:04:05")
latencyStr := strconv.FormatInt(latency, 10) + " us"
srcInodeStr := strconv.FormatUint(srcInode, 10)
dstInodeStr := strconv.FormatUint(dstInode, 10)
entry = fmt.Sprintf("%s, %s, %s, %s, %s, %s, %s, %s, %s, %s",
curTimeStr, ipAddr, hostName, op, src, dst, errStr, latencyStr, srcInodeStr, dstInodeStr)
return
}
func FormatLog(op, src, dst string, err error, latency int64, srcInode, dstInode uint64) {
if entry := formatAuditEntry(op, src, dst, err, latency, srcInode, dstInode); entry != "" {
AddLog(entry)
}
}
func ResetWriterBufferSize(size int) {
AdtMutex.Lock()
defer AdtMutex.Unlock()
if gAdt == nil {
return
}
gAdt.resetWriterBuffC <- size
}
func AddLog(content string) {
AdtMutex.Lock()
defer AdtMutex.Unlock()
if gAdt == nil {
return
}
select {
case gAdt.bufferC <- content:
return
default:
log.LogErrorf("Async audit log failed, audit:[%s]", content)
}
}
func StopAudit() {
AdtMutex.Lock()
defer AdtMutex.Unlock()
if gAdt == nil {
return
}
gAdt.stop()
gAdt = nil
}
func (a *Audit) flushAuditLog() {
cleanTimer := time.NewTimer(DefaultCleanInterval)
for {
select {
case <-a.stopC:
return
case bufSize := <-a.resetWriterBuffC:
a.writerBufSize = bufSize
a.newWriterSize(bufSize)
case aLog := <-a.bufferC:
a.logAudit(aLog)
case <-cleanTimer.C:
a.removeLogFile()
cleanTimer.Reset(DefaultCleanInterval)
}
}
}
func (a *Audit) newWriterSize(size int) error {
a.writerBufSize = size
if a.writer != nil {
a.writer.Flush()
}
if a.logFile == nil {
logFile, err := os.OpenFile(a.logFileName, os.O_RDWR|os.O_APPEND|os.O_CREATE, 0666)
if err != nil {
log.LogErrorf("newWriterSize failed, logFileName: %s, err: %v\n", a.logFileName, err)
return fmt.Errorf("OpenLogFile failed, logFileName %s\n", a.logFileName)
}
a.logFile = logFile
if size <= 0 {
log.LogErrorf("newWriterSize : buffer for logFileName is disabled")
a.writer = bufio.NewWriter(logFile)
} else {
a.writer = bufio.NewWriterSize(logFile, size)
}
} else {
_, err := a.logFile.Stat()
if err == nil {
if size <= 0 {
log.LogErrorf("newWriterSize : buffer for logFileName is disabled")
a.writer = bufio.NewWriter(a.logFile)
} else {
a.writer = bufio.NewWriterSize(a.logFile, size)
}
} else {
a.logFile.Close()
a.logFile = nil
return a.newWriterSize(size)
}
}
return nil
}
func (a *Audit) removeLogFile() {
fs := syscall.Statfs_t{}
if err := syscall.Statfs(a.logDir, &fs); err != nil {
log.LogErrorf("Get fs stat failed, err: %v", err)
return
}
diskSpaceLeft := int64(fs.Bavail * uint64(fs.Bsize))
diskSpaceLeft -= DefaultHeadRoom * 1024 * 1024
fInfos, err := ioutil.ReadDir(a.logDir)
if err != nil {
log.LogErrorf("ReadDir failed, logDir: %s, err: %v", a.logDir, err)
return
}
var needDelFiles ShiftedFile
for _, info := range fInfos {
if a.shouldDelete(info, diskSpaceLeft, Audit_Module) {
needDelFiles = append(needDelFiles, info)
}
}
sort.Sort(needDelFiles)
for _, info := range needDelFiles {
if err = os.Remove(path.Join(a.logDir, info.Name())); err != nil {
log.LogErrorf("Remove log file failed, logFileName: %s, err: %v", info.Name(), err)
continue
}
diskSpaceLeft += info.Size()
if diskSpaceLeft > 0 && time.Since(info.ModTime()) < MaxReservedDays {
break
}
}
}
func (a *Audit) shouldDelete(info os.FileInfo, diskSpaceLeft int64, module string) bool {
isOldAuditLogFile := info.Mode().IsRegular() && strings.HasSuffix(info.Name(), ShiftedExtension) && strings.HasPrefix(info.Name(), module)
if diskSpaceLeft <= 0 {
return isOldAuditLogFile
}
return time.Since(info.ModTime()) > MaxReservedDays && isOldAuditLogFile
}
func (a *Audit) stop() {
close(a.stopC)
a.writer.Flush()
a.logFile.Close()
}
func (a *Audit) logAudit(content string) error {
a.shiftFiles()
fmt.Fprintf(a.writer, "%s\n", content)
if a.writerBufSize <= 0 {
a.writer.Flush()
}
return nil
}
func (a *Audit) shiftFiles() error {
fileInfo, err := os.Stat(a.logFileName)
if err != nil {
return err
}
if fileInfo.Size() < a.logMaxSize {
return nil
}
if syscall.Access(a.logFileName, F_OK) == nil {
logNewFileName := a.logFileName + "." + time.Now().Format(FileNameDateFormat) + ShiftedExtension
a.writer.Flush()
a.logFile.Close()
a.writer = nil
a.logFile = nil
if os.Rename(a.logFileName, logNewFileName) != nil {
log.LogErrorf("RenameFile failed, logFileName: %s, logNewFileName: %s, err: %v\n",
a.logFileName, logNewFileName, err)
return fmt.Errorf("RenameFile failed, logFileName %s, logNewFileName %s\n",
a.logFileName, logNewFileName)
}
}
return a.newWriterSize(a.writerBufSize)
}