1. Add container router

2. Implement sysdig server

Author: YLonely

go.mod

@@ -1,4 +1,4 @@
-module sysdig-monitor
+module github.com/YLonely/sysdig-monitor
 
 go 1.12
 

server/router/container/container.go

@@ -0,0 +1,26 @@
+package container
+
+import (
+	"github.com/YLonely/sysdig-monitor/server/router"
+)
+
+type containerRouter struct {
+	routes []router.Route
+	// sysdig client
+
+	// docker client
+}
+
+func NewRouter() router.Router {
+	r := &containerRouter{}
+
+	return r
+}
+
+func (r *containerRouter) Routes() []router.Route {
+	return r.routes
+}
+
+func (r *containerRouter) initRoutes() {
+	r.routes = []router.Route{}
+}

server/router/router.go

@@ -8,7 +8,7 @@ type HandlerType func(c *gin.Context)
 
 // Router defines an interface to specify a group of routes ot add to the server
 type Router interface {
-	Routes()
+	Routes() []Route
 }
 
 // Route defines an individual API route in the sysdig-monitor server

sysdig/event.go

@@ -0,0 +1,30 @@
+package sysdig
+
+// Event represents a container event in sysdig
+type Event struct {
+	ID              string `json:"container.id"`
+	Name            string `json:"container.name"`
+	CPUId           int    `json:"evt.cpu"`
+	Dir             string `json:"evt.dir"`
+	Info            string `json:"evt.info"`
+	Num             int    `json:"evt.num"`
+	TimeStamp       int64  `json:"evt.outputtime"`
+	Type            string `json:"evt.type"`
+	ProcName        string `json:"proc.name"`
+	ThreadID        int    `json:"thread.tid"`
+	ThreadVirtualID int    `json:"thread.vid"`
+}
+
+type subscriber struct {
+	c      chan Event
+	closed bool
+}
+
+func (s *subscriber) close() {
+	if s.closed {
+		return
+	}
+	close(s.c)
+	s.closed = true
+	return
+}

sysdig/monitor.go

@@ -0,0 +1,55 @@
+package sysdig
+
+import (
+	"os/exec"
+	"syscall"
+	"time"
+)
+
+var Monitor ProcessMonitor = &defaultMonitor{}
+
+// Exit info
+type Exit struct {
+	TimeStamp time.Time
+	Pid       int
+	Status    int
+}
+
+// ProcessMonitor is an interface for monitoring process
+type ProcessMonitor interface {
+	Start(*exec.Cmd) (chan Exit, error)
+	Wait(*exec.Cmd, chan Exit) (int, error)
+}
+
+type defaultMonitor struct {
+}
+
+func (m *defaultMonitor) Start(c *exec.Cmd) (chan Exit, error) {
+	if err := c.Start(); err != nil {
+		return nil, err
+	}
+	ec := make(chan Exit, 1)
+	go func() {
+		var status int
+		if err := c.Wait(); err != nil {
+			status = 255
+			if exitErr, ok := err.(*exec.ExitError); ok {
+				if ws, ok := exitErr.Sys().(syscall.WaitStatus); ok {
+					status = ws.ExitStatus()
+				}
+			}
+		}
+		ec <- Exit{
+			TimeStamp: time.Now(),
+			Pid:       c.Process.Pid,
+			Status:    status,
+		}
+		close(ec)
+	}()
+	return ec, nil
+}
+
+func (m *defaultMonitor) Wait(c *exec.Cmd, ec chan Exit) (int, error) {
+	e := <-ec
+	return e.Status, nil
+}

sysdig/sysdig.go

@@ -0,0 +1,100 @@
+package sysdig
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"os/exec"
+)
+
+const binaryName = "sysdig"
+const bufferSize = 1024
+
+// Server starts sysdig and dispatch events
+type Server interface {
+	Subscribe() chan Event
+	Start(ctx context.Context) error
+}
+
+var _ Server = &localServer{}
+
+type localServer struct {
+	subscribers []*subscriber
+}
+
+// NewServer creates a server
+func NewServer() Server {
+	return &localServer{}
+}
+
+func (ls *localServer) Subscribe() chan Event {
+	c := make(chan Event, bufferSize)
+	ls.subscribers = append(ls.subscribers, &subscriber{c: c})
+	return c
+}
+
+func (ls *localServer) Start(ctx context.Context) error {
+	if err := ls.preRrequestCheck(); err != nil {
+		return err
+	}
+	args := []string{"-pc", "-j", "container.id!=host"}
+	cmd := exec.CommandContext(ctx, binaryName, args...)
+	rd, err := cmd.StdoutPipe()
+	if err != nil {
+		return err
+	}
+	ec, err := Monitor.Start(cmd)
+	if err != nil {
+		rd.Close()
+		return err
+	}
+
+	var (
+		dec = json.NewDecoder(rd)
+	)
+
+	go func() {
+		defer func() {
+			rd.Close()
+			Monitor.Wait(cmd, ec)
+		}()
+		for {
+			var e Event
+			if err := dec.Decode(&e); err != nil {
+				if err == io.EOF {
+					return
+				}
+				e = Event{Type: "error"}
+			}
+			for _, subscriber := range ls.subscribers {
+				if !subscriber.closed {
+					subscriber.c <- e
+				}
+			}
+		}
+	}()
+
+	return nil
+}
+
+func (ls *localServer) preRrequestCheck() error {
+	//try run sysdig
+	ctx, cancel := context.WithCancel(context.Background())
+	cmd := exec.CommandContext(ctx, binaryName)
+	var (
+		ec  chan Exit
+		err error
+	)
+
+	if ec, err = Monitor.Start(cmd); err != nil {
+		cancel()
+		return errors.New("can not start sysdig")
+	}
+
+	cancel()
+
+	Monitor.Wait(cmd, ec)
+
+	return nil
+}