Expand connector buffer, put all file info into db

This commit I make several changes, reasons are as follows:
- Expand netlink connector buffer size. Originally it uses only one
  page of mem, but sometimes it causes "no enough buffer" errno, or
  the socket is blocked long time. Now it's 20 pages.
- All file infos are thrown into database. As the last commit co-
  mment, There's 2 tables, "fds" and "files". When a file discriptor
  is closed, the info in fds will be found, delete, and put into the
  "file" table with its close time.

Left questions:
- The netlink connector is always found blocked without any reasons.
  Fix it, or replace the golang-coded connector with C program? The
  key is why it's blocked. Maybe it's in the kernel src code.
- sometimes audit still losts info(not too much). For instance, I
  use vim in the docker to change hello.c, the hello.c may be opened
  but no close info recvd. Or, the swap file of vim, such as
  .hello.c.swp or .hello.c.swx is not closed. What's more, the hello.c
  is never written, but swap files are. May vim write to swap files,
  and replace the origin file? Let's check it.
- Besides, when a pid exits, we should check its file discriptors
  and close them all.

5 files changed, 45 insertions, 33 deletions

diff --git a/src/deal.go b/src/deal.go
index 483d4d2..1dd309c 100644
--- a/src/deal.go
+++ b/src/deal.go
@@ -67,17 +67,19 @@ func deal() {
                 case PIDEXIT:
                         go deletePid(cooked)
                 case FILEOPEN:
-                        fileOpen(cooked)
+                        go fileOpen(cooked)
                 case FILEWRITE:
-                        fileWrite(cooked)
+                        go fileWrite(cooked)
                 case FILECLOSE:
-                        fileClose(cooked)
+                        go fileClose(cooked)
                 }
         }
 }
 
 func deletePid(cooked Event) {
-        pidCol.GetLock()
+        if !pidCol.GetLock() {
+                return
+        }
         // 先从老爹那里销户
         pidCol.UpdateOne(bson.M{"pid": cooked.ppid}, bson.M{
                 "$pull": bson.M{
@@ -102,9 +104,9 @@ func deletePid(cooked Event) {
 }
 
 func dealNewPid(cooked Event) {
-        fmt.Printf("Fork: %v\t%6d\t%6d\n", cooked.timestamp, cooked.ppid, cooked.pid)
+        fmt.Printf("Fork\t%6d\t%6d\t%6d\t%6d\n", cooked.ppid, cooked.parentTgid, cooked.pid, cooked.tgid)
         // 有无父进程在观察中
-        docRes, err = pidCol.Finddoc(bson.M{"pid": cooked.ppid})
+        docRes, err = pidCol.Finddoc(bson.M{"pid": cooked.parentTgid})
         if err != nil || len(docRes) != 1 {
                 return
         }
@@ -116,7 +118,9 @@ func dealNewPid(cooked Event) {
                 return
         }
 
-        pidCol.GetLock()
+        if !pidCol.GetLock() {
+                return
+        }
         if len(docRes) != 0 {
                 // 进程原本就存在，换言之别的消息先到了
                 // 所有先行抵达的消息必须保留execve/children字段
@@ -125,7 +129,9 @@ func dealNewPid(cooked Event) {
                 pidCol.UpdateOne(bson.M{"pid": cooked.pid}, bson.M{
                         "start_timestamp": cooked.timestamp,
                         "ppid":            cooked.ppid,
+                        "parentTgid":      cooked.parentTgid,
                         "pid":             cooked.pid,
+                        "tgid":            cooked.tgid,
                         "cwd":             cooked.cwd,
                         // "execve":          []bson.M{},
                         "args": cooked.argv,
@@ -136,7 +142,9 @@ func dealNewPid(cooked Event) {
                 pidCol.InsertOne(bson.M{
                         "start_timestamp": cooked.timestamp,
                         "ppid":            cooked.ppid,
+                        "parentTgid":      cooked.parentTgid,
                         "pid":             cooked.pid,
+                        "tgid":            cooked.tgid,
                         "cwd":             cooked.cwd,
                         "execve":          []bson.M{},
                         "args":            cooked.argv,
@@ -153,7 +161,6 @@ func dealNewPid(cooked Event) {
 }
 
 func dealExecve(cooked Event) {
-        fmt.Printf("EXEC: %6d\t%6d\n", cooked.ppid, cooked.pid)
         // 父进程在不在？不在扔
         docRes, err = pidCol.Finddoc(bson.M{"pid": cooked.ppid})
         if err != nil || len(docRes) != 1 {
@@ -166,7 +173,9 @@ func dealExecve(cooked Event) {
                 return
         }
 
-        pidCol.GetLock()
+        if !pidCol.GetLock() {
+                return
+        }
         if len(docRes) == 1 {
                 // 自身已在，直接记录
                 pidCol.UpdateOne(bson.M{"pid": cooked.pid}, bson.M{
@@ -215,22 +224,14 @@ func fileOpen(cooked Event) {
 }
 
 func fileClose(cooked Event) {
-        res, err := fdCol.Finddoc(bson.M{
-                "pid":             cooked.pid,
-                "fd":              cooked.syscallParam[0],
-                "close_timestamp": bson.M{"$exists": false},
-        })
+        res, err := fdCol.FindOneAndDelete(bson.M{"pid": cooked.pid, "fd": cooked.syscallParam[0]})
         if err != nil {
-                fmt.Printf("Err closing fd %d of pid %d: %v\n", cooked.syscallParam[0], cooked.pid, err)
-        }
-        if len(res) == 0 {
                 return
         }
-        fdCol.UpdateOne(bson.M{
-                "pid":             cooked.pid,
-                "fd":              cooked.syscallParam[0],
-                "close_timestamp": bson.M{"$exists": false},
-        }, bson.M{"$set": bson.M{"close_timestamp": cooked.timestamp}})
+        res["close_timestamp"] = cooked.timestamp
+        if err := fileCol.InsertOne(res); err != nil {
+                fmt.Printf("Err inserting files: %v\n", err)
+        }
 }
 
 func fileWrite(cooked Event) {
diff --git a/src/godo.go b/src/godo.go
index 77e677c..923ef85 100644
--- a/src/godo.go
+++ b/src/godo.go
@@ -44,14 +44,14 @@ func main() {
         var auditCmd *exec.Cmd
 
         pidSyscall := []string{"execve"}
-        // 设置监听规则
+        // // 设置监听规则
         for i := 0; i < len(pidSyscall); i++ {
                 auditCmd = exec.Command("auditctl", "-a", "exit,always", "-F", "arch=b64", "-S", pidSyscall[i])
                 auditCmd.Run()
         }
 
         // 监听文件的消息
-        fileSyscall := []string{"open", "write", "close"}
+        fileSyscall := []string{"open", "close", "write"}
         // fileSyscall := []string{"open", "write", "creat", "unlink", "opendir", "mkdir", "rmdir", "chmod", "fchmod", "chown", "fchown", "lchown", "flock"}
         for i := 0; i < len(fileSyscall); i++ {
                 auditCmd = exec.Command("auditctl", "-a", "exit,always", "-F", "arch=b64", "-S", fileSyscall[i])
@@ -106,7 +106,7 @@ func procWatch() error {
         }
         defer ns.Close()
         for {
-                res, err := ns.Receive()
+                res, err := ns.Receive(20)
                 if err != nil {
                         fmt.Printf("Error recv: %v\n", err)
                         continue
diff --git a/src/mongo.go b/src/mongo.go
index 3a23131..764f877 100644
--- a/src/mongo.go
+++ b/src/mongo.go
@@ -78,6 +78,13 @@ func (mc *mongoClient) Finddoc(filter bson.M) ([]bson.M, error) {
         return results, err
 }
 
+func (mc *mongoClient) FindOneAndDelete(filter bson.M) (bson.M, error) {
+        res := mc.col.FindOneAndDelete(context.Background(), filter)
+        var result bson.M
+        err := res.Decode(&result)
+        return result, err
+}
+
 func (mc *mongoClient) Drop() error {
         return mc.col.Drop(context.Background())
 }
@@ -94,11 +101,14 @@ func (mc *mongoClient) Disconnect() error {
         return nil
 }
 
-func (mc *mongoClient) GetLock() {
-        for i := 0; i < 20000; {
+func (mc *mongoClient) GetLock() bool {
+        for i := 0; i < 200000; {
                 if !mc.Mutex.TryLock() {
                         i++
+                } else {
+                        return true
                 }
         }
         fmt.Printf("Die...\n")
+        return false
 }
diff --git a/src/netlink b/src/netlink
-Subproject 10d3ea361f0bcafb9b92054440984d32421aeb7
+Subproject e53c2724725c5991cdd9ea088c26832c5c9fcf0
diff --git a/src/organize.go b/src/organize.go
index 238509f..5268a90 100644
--- a/src/organize.go
+++ b/src/organize.go
@@ -47,20 +47,21 @@ func orgnaze() {
                         break
                 }
                 rawEvent = raw.(libaudit.RawAuditMessage)
+                fmt.Printf("type=%v msg=%s\n", rawEvent.Type, rawEvent.Data)
 
                 switch rawEvent.Type {
                 case auparse.AUDIT_SYSCALL:
-                        syscallRaw(rawEvent)
+                        go syscallRaw(rawEvent)
                 case auparse.AUDIT_EXECVE:
-                        execve(rawEvent)
+                        go execve(rawEvent)
                 case auparse.AUDIT_CWD:
-                        cwd(rawEvent)
+                        go cwd(rawEvent)
                 case auparse.AUDIT_PATH:
-                        path(rawEvent)
+                        go path(rawEvent)
                 case auparse.AUDIT_PROCTITLE:
-                        proctitle(rawEvent)
+                        go proctitle(rawEvent)
                 case auparse.AUDIT_EOE:
-                        eoe(rawEvent)
+                        go eoe(rawEvent)
                 default:
                         // ATTENTION: 这里也需要做防护
                 }