diff --git a/experimental/example/Makefile b/experimental/example/Makefile index ca854e597..00b8085e8 100644 --- a/experimental/example/Makefile +++ b/experimental/example/Makefile @@ -1,7 +1,7 @@ CC=gcc CFLAGS=-g -lpthread -all: qotd.marker app.marker client.marker searchapp.marker shout.marker frontend.marker +all: qotd.marker app.marker client.marker searchapp.marker shout.marker frontend.marker echo.marker qotd/qotd: qotd/qotd.o gcc -o $@ $< $(CFLAGS) @@ -21,6 +21,7 @@ client.marker: client/* searchapp.marker: searchapp/* searchapp/searchapp shout.marker: shout/* shout/shout frontend.marker: frontend/* +echo.marker: echo/* %.marker: docker build -t tomwilkie/$( +WORKDIR /home/weave +ADD requirements.txt /home/weave/ +RUN pip install -r /home/weave/requirements.txt +ADD echo.py /home/weave/ +EXPOSE 5000 +ENTRYPOINT ["python", "/home/weave/echo.py"] diff --git a/experimental/example/echo/echo.py b/experimental/example/echo/echo.py new file mode 100644 index 000000000..68fd95ab6 --- /dev/null +++ b/experimental/example/echo/echo.py @@ -0,0 +1,27 @@ +import os +import socket +import sys +import random +import time +import threading +import logging + + +from flask import Flask +from flask import request +from werkzeug.serving import WSGIRequestHandler + +app = Flask(__name__) + +@app.route('/') +def hello(): + if random.random() > 0.6: + time.sleep(2) + else: + time.sleep(0.5) + return request.data + +if __name__ == "__main__": + logging.basicConfig(format='%(asctime)s %(levelname)s %(filename)s:%(lineno)d - %(message)s', level=logging.INFO) + WSGIRequestHandler.protocol_version = "HTTP/1.0" + app.run(host="0.0.0.0", port=80, debug=True) diff --git a/experimental/example/echo/requirements.txt b/experimental/example/echo/requirements.txt new file mode 100644 index 000000000..cd204415b --- /dev/null +++ b/experimental/example/echo/requirements.txt @@ -0,0 +1,2 @@ +flask + diff --git a/experimental/example/qotd/qotd.c b/experimental/example/qotd/qotd.c index d61314966..082f99eaa 100644 --- a/experimental/example/qotd/qotd.c +++ b/experimental/example/qotd/qotd.c @@ -8,17 +8,13 @@ /* this function is run by the thread */ void *thread_func(void *sock) { struct sockaddr_in dest; - char buffer[1024]; + char in_buffer[1024]; + char out_buffer[1024]; int sockfd = (int)sock; int clientfd; printf("I'm thread %d\n", syscall(SYS_gettid)); - if (read(sockfd, buffer, sizeof(buffer)) < 0) { - perror("ERROR reading from socket"); - return; - } - clientfd = socket(AF_INET, SOCK_STREAM, 0); if (clientfd < 0) { perror("ERROR opening socket"); @@ -37,10 +33,11 @@ void *thread_func(void *sock) { return; } - int readbytes = recv(clientfd, buffer, sizeof(buffer), 0); + int readbytes = recv(clientfd, in_buffer, sizeof(in_buffer), 0); close(clientfd); - if (write(sockfd, buffer, read) < 0) { + int writtenbytes = snprintf(out_buffer, sizeof(out_buffer), "{\"qotd\": %s}\n", in_buffer); + if (write(sockfd, out_buffer, writtenbytes) < 0) { perror("ERROR writing to socket"); return; } diff --git a/experimental/example/run.sh b/experimental/example/run.sh index ceeac18ca..8be48875e 100755 --- a/experimental/example/run.sh +++ b/experimental/example/run.sh @@ -23,6 +23,8 @@ start_container 1 elasticsearch elasticsearch start_container 2 tomwilkie/searchapp searchapp start_container 1 redis redis start_container 1 tomwilkie/qotd qotd +start_container 1 tomwilkie/echo echo start_container 2 tomwilkie/app app start_container 2 tomwilkie/frontend frontend --add-host=dns.weave.local:$(weave docker-bridge-ip) start_container 1 tomwilkie/client client + diff --git a/experimental/tracer/.gitignore b/experimental/tracer/.gitignore new file mode 100644 index 000000000..681fc6b1d --- /dev/null +++ b/experimental/tracer/.gitignore @@ -0,0 +1,3 @@ +main/main +tracer.tar +main/static.go diff --git a/experimental/tracer/Makefile b/experimental/tracer/Makefile new file mode 100644 index 000000000..e66b58668 --- /dev/null +++ b/experimental/tracer/Makefile @@ -0,0 +1,14 @@ +tracer.tar: main/main main/Dockerfile + docker build -t weaveworks/tracer main/ + docker save weaveworks/tracer:latest >$@ + +main/main: main/*.go main/static.go ptrace/*.go + go get -tags netgo ./$(@D) + go build -ldflags "-extldflags \"-static\"" -tags netgo -o $@ ./$(@D) + +main/static.go: ui/* + esc -o main/static.go -prefix ui ui + +clean: + go clean ./.. + rm -f main/static.go tracer.tar main/main diff --git a/experimental/tracer/README.md b/experimental/tracer/README.md new file mode 100644 index 000000000..23cc29f33 --- /dev/null +++ b/experimental/tracer/README.md @@ -0,0 +1,11 @@ +Run tracer: +- make +- ./tracer.sh start + +TODO: +- need to stich traces together +- deal with persistant connections +- make it work for goroutines +- test with jvm based app +- find way to get local ip address for an incoming connection +- make the container/process trace start/stop more reliable diff --git a/experimental/tracer/main/Dockerfile b/experimental/tracer/main/Dockerfile new file mode 100644 index 000000000..c9555cb51 --- /dev/null +++ b/experimental/tracer/main/Dockerfile @@ -0,0 +1,6 @@ +FROM gliderlabs/alpine +MAINTAINER Weaveworks Inc +WORKDIR /home/weave +COPY ./main /home/weave/ +EXPOSE 4050 +ENTRYPOINT ["/home/weave/main"] diff --git a/experimental/tracer/main/http.go b/experimental/tracer/main/http.go new file mode 100644 index 000000000..4029bb573 --- /dev/null +++ b/experimental/tracer/main/http.go @@ -0,0 +1,150 @@ +package main + +import ( + "encoding/json" + "fmt" + "log" + "mime" + "net/http" + "strconv" + "strings" + + "github.com/gorilla/mux" + + "github.com/weaveworks/scope/probe/docker" + "github.com/weaveworks/scope/probe/process" +) + +func respondWith(w http.ResponseWriter, code int, response interface{}) { + w.Header().Set("Content-Type", "application/json") + w.Header().Add("Cache-Control", "no-cache") + w.WriteHeader(code) + if err := json.NewEncoder(w).Encode(response); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + log.Printf("Error handling http request: %v", err.Error()) + } +} + +func (t *tracer) pidsForContainer(id string) ([]int, error) { + var container docker.Container + t.docker.WalkContainers(func(c docker.Container) { + if c.ID() == id { + container = c + } + }) + + if container == nil { + return []int{}, fmt.Errorf("Not Found") + } + + pidTree, err := process.NewTree(process.NewWalker("/proc")) + if err != nil { + return []int{}, err + } + + return pidTree.GetChildren(container.PID()) +} + +// Container is the type exported by the HTTP API. +type Container struct { + ID string + Name string + PIDs []int +} + +func (t *tracer) http(port int) { + router := mux.NewRouter() + + router.Methods("GET").Path("/container").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + pidTree, err := process.NewTree(process.NewWalker("/proc")) + if err != nil { + respondWith(w, http.StatusBadRequest, err.Error()) + return + } + + containers := []Container{} + t.docker.WalkContainers(func(container docker.Container) { + children, _ := pidTree.GetChildren(container.PID()) + out := Container{ + Name: strings.TrimPrefix(container.Container().Name, "/"), + ID: container.ID(), + PIDs: children, + } + containers = append(containers, out) + }) + + respondWith(w, http.StatusOK, containers) + }) + + router.Methods("POST").Path("/container/{id}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + id := mux.Vars(r)["id"] + children, err := t.pidsForContainer(id) + if err != nil { + respondWith(w, http.StatusBadRequest, err.Error()) + return + } + + for _, pid := range children { + t.ptrace.TraceProcess(pid) + } + w.WriteHeader(204) + }) + + router.Methods("DELETE").Path("/container/{id}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + id := mux.Vars(r)["id"] + children, err := t.pidsForContainer(id) + if err != nil { + respondWith(w, http.StatusBadRequest, err.Error()) + return + } + + for _, pid := range children { + t.ptrace.StopTracing(pid) + } + w.WriteHeader(204) + }) + + router.Methods("GET").Path("/pid").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + respondWith(w, http.StatusOK, t.ptrace.AttachedPIDs()) + }) + + router.Methods("POST").Path("/pid/{pid:\\d+}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + pid, err := strconv.Atoi(mux.Vars(r)["pid"]) + if err != nil { + respondWith(w, http.StatusBadRequest, err.Error()) + return + } + + t.ptrace.TraceProcess(pid) + w.WriteHeader(204) + }) + + router.Methods("DELETE").Path("/pid/{pid:\\d+}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + pid, err := strconv.Atoi(mux.Vars(r)["pid"]) + if err != nil { + respondWith(w, http.StatusBadRequest, err.Error()) + return + } + + t.ptrace.StopTracing(pid) + w.WriteHeader(204) + }) + + router.Methods("GET").Path("/traces").HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + respondWith(w, http.StatusOK, t.store.Traces()) + }) + + mime.AddExtensionType(".svg", "image/svg+xml") + + router.Methods("GET").PathPrefix("/").Handler(http.FileServer(FS(false))) // everything else is static + + log.Printf("Launching HTTP API on port %d", port) + srv := &http.Server{ + Addr: fmt.Sprintf(":%d", port), + Handler: router, + } + + if err := srv.ListenAndServe(); err != nil { + log.Printf("Unable to create http listener: %v", err) + } +} diff --git a/experimental/tracer/main/main.go b/experimental/tracer/main/main.go new file mode 100644 index 000000000..15120a0cb --- /dev/null +++ b/experimental/tracer/main/main.go @@ -0,0 +1,70 @@ +package main + +import ( + "log" + _ "net/http/pprof" + "os" + "os/signal" + "runtime" + "syscall" + "time" + + "github.com/weaveworks/scope/experimental/tracer/ptrace" + "github.com/weaveworks/scope/probe/docker" +) + +const ( + procRoot = "/proc" + pollInterval = 10 * time.Second +) + +type tracer struct { + ptrace ptrace.PTracer + store *store + docker docker.Registry +} + +func (t *tracer) Stop() { + log.Printf("Shutting down...") + t.ptrace.Stop() + t.docker.Stop() + log.Printf("Done.") +} + +func main() { + dockerRegistry, err := docker.NewRegistry(pollInterval, nil) + if err != nil { + log.Fatalf("Could start docker watcher: %v", err) + } + + store := newStore() + tracer := tracer{ + store: store, + ptrace: ptrace.NewPTracer(store), + docker: dockerRegistry, + } + defer tracer.Stop() + + go tracer.http(6060) + <-handleSignals() +} + +func handleSignals() chan struct{} { + quit := make(chan struct{}, 10) + go func() { + sigs := make(chan os.Signal, 1) + signal.Notify(sigs, syscall.SIGQUIT, syscall.SIGINT, syscall.SIGTERM) + buf := make([]byte, 1<<20) + for { + sig := <-sigs + switch sig { + case syscall.SIGINT, syscall.SIGTERM: + quit <- struct{}{} + case syscall.SIGQUIT: + stacklen := runtime.Stack(buf, true) + log.Printf("=== received SIGQUIT ===\n*** goroutine dump...\n%s\n*** end\n", buf[:stacklen]) + } + } + }() + return quit +} diff --git a/experimental/tracer/main/store.go b/experimental/tracer/main/store.go new file mode 100644 index 000000000..5b4639360 --- /dev/null +++ b/experimental/tracer/main/store.go @@ -0,0 +1,172 @@ +package main + +import ( + "fmt" + "math/rand" + "sort" + "sync" + + "github.com/msackman/skiplist" + + "github.com/weaveworks/scope/experimental/tracer/ptrace" +) + +const epsilon = int64(5) * 1000 // milliseconds + +// Traces are indexed by from addr, from port, and start time. +type key struct { + fromAddr uint32 + fromPort uint16 + startTime int64 +} + +func (k key) MarshalJSON() ([]byte, error) { + return []byte(fmt.Sprintf("\"%x.%x.%x\"", k.fromAddr, k.fromPort, k.startTime)), nil +} + +type trace struct { + PID int + Key key + ServerDetails *ptrace.ConnectionDetails + ClientDetails *ptrace.ConnectionDetails + Children []*trace + Level int +} + +type byKey []*trace + +func (a byKey) Len() int { return len(a) } +func (a byKey) Swap(i, j int) { a[i], a[j] = a[j], a[i] } +func (a byKey) Less(i, j int) bool { return a[i].Key.startTime < a[j].Key.startTime } + +type store struct { + sync.RWMutex + traces *skiplist.SkipList +} + +func newKey(fd *ptrace.Fd) key { + var fromAddr uint32 + for _, b := range fd.FromAddr.To4() { + fromAddr <<= 8 + fromAddr |= uint32(b) + } + + return key{fromAddr, fd.FromPort, fd.Start} +} + +func (k key) LessThan(other skiplist.Comparable) bool { + r := other.(key) + + if k.fromAddr != r.fromAddr { + return k.fromAddr > r.fromAddr + } + + if k.fromPort != r.fromPort { + return k.fromPort < r.fromPort + } + + if k.Equal(other) { + return false + } + + return k.startTime < r.startTime +} + +func (k key) Equal(other skiplist.Comparable) bool { + r := other.(key) + if k.fromAddr != r.fromAddr || k.fromPort != r.fromPort { + return false + } + + diff := k.startTime - r.startTime + return -epsilon < diff && diff < epsilon +} + +func newStore() *store { + return &store{traces: skiplist.New(rand.New(rand.NewSource(0)))} +} + +func (t *trace) addChild(child *trace) { + // find the child we're supposed to be replacing + for i, candidate := range t.Children { + if !candidate.Key.Equal(skiplist.Comparable(child.Key)) { + continue + } + + // Fix up some fields + child.ClientDetails = candidate.ClientDetails + IncrementLevel(child, t.Level+1) + + // Overwrite old record + t.Children[i] = child + return + } +} + +func (s *store) RecordConnection(pid int, connection *ptrace.Fd) { + s.Lock() + defer s.Unlock() + + newTrace := &trace{ + PID: pid, + Key: newKey(connection), + ServerDetails: &connection.ConnectionDetails, + } + for _, child := range connection.Children { + newTrace.Children = append(newTrace.Children, &trace{ + Level: 1, + Key: newKey(child), + ClientDetails: &child.ConnectionDetails, + }) + } + + // First, see if this new conneciton is a child of an existing connection. + // This indicates we have a parent connection to attach to. + // If not, insert this connection. + if parentNode := s.traces.Get(newTrace.Key); parentNode != nil { + parentTrace := parentNode.Value.(*trace) + parentTrace.addChild(newTrace) + parentNode.Remove() + } else { + s.traces.Insert(newTrace.Key, newTrace) + } + + // Next, see if we already know about the child connections + // If not, insert each of our children. + for _, child := range newTrace.Children { + if childNode := s.traces.Get(child.Key); childNode != nil { + childTrace := childNode.Value.(*trace) + newTrace.addChild(childTrace) + childNode.Remove() + } else { + s.traces.Insert(child.Key, newTrace) + } + } +} + +// IncrementLevel ... +func IncrementLevel(trace *trace, increment int) { + trace.Level += increment + for _, child := range trace.Children { + IncrementLevel(child, increment) + } +} + +func (s *store) Traces() []*trace { + s.RLock() + defer s.RUnlock() + + traces := []*trace{} + var cur = s.traces.First() + for cur != nil { + key := cur.Key.(key) + trace := cur.Value.(*trace) + if trace.Key == key { + traces = append(traces, trace) + } + cur = cur.Next() + } + + sort.Sort(byKey(traces)) + return traces +} diff --git a/experimental/tracer/ptrace/fd.go b/experimental/tracer/ptrace/fd.go new file mode 100644 index 000000000..463a35628 --- /dev/null +++ b/experimental/tracer/ptrace/fd.go @@ -0,0 +1,198 @@ +package ptrace + +import ( + "bufio" + "encoding/hex" + "fmt" + "net" + "os" + "regexp" + "strconv" + "time" +) + +const ( + listening = iota + incoming + outgoing +) + +const ( + socketPattern = `^socket:\[(\d+)\]$` + tcpPattern = `^\s*(?P\d+): (?P[A-F0-9]{8}):(?P[A-F0-9]{4}) ` + + `(?P[A-F0-9]{8}):(?P[A-F0-9]{4}) (?:[A-F0-9]{2}) (?:[A-F0-9]{8}):(?:[A-F0-9]{8}) ` + + `(?:[A-F0-9]{2}):(?:[A-F0-9]{8}) (?:[A-F0-9]{8}) \s+(?:\d+) \s+(?:\d+) (?P\d+)` +) + +var ( + socketRegex = regexp.MustCompile(socketPattern) + tcpRegexp = regexp.MustCompile(tcpPattern) +) + +// ConnectionDetails ... +type ConnectionDetails struct { + direction int + + Start int64 + Stop int64 + sent int64 + received int64 + + FromAddr net.IP + FromPort uint16 + ToAddr net.IP + ToPort uint16 +} + +// Fd represents a connect and subsequent connections caused by it. +type Fd struct { + fd int + closed bool + + ConnectionDetails + + // Fds are connections, and can have a causal-link to other Fds + Children []*Fd +} + +func getLocalAddr(pid, fd int) (addr net.IP, port uint16, err error) { + var ( + socket string + match []string + inode int + tcpFile *os.File + scanner *bufio.Scanner + candidate int + port64 int64 + ) + + socket, err = os.Readlink(fmt.Sprintf("/proc/%d/fd/%d", pid, fd)) + if err != nil { + return + } + + match = socketRegex.FindStringSubmatch(socket) + if match == nil { + err = fmt.Errorf("Fd %d not a socket", fd) + return + } + + inode, err = strconv.Atoi(match[1]) + if err != nil { + return + } + + tcpFile, err = os.Open(fmt.Sprintf("/proc/%d/net/tcp", pid)) + if err != nil { + return + } + defer tcpFile.Close() + + scanner = bufio.NewScanner(tcpFile) + for scanner.Scan() { + match = tcpRegexp.FindStringSubmatch(scanner.Text()) + if match == nil { + continue + } + + candidate, err = strconv.Atoi(match[6]) + if err != nil { + return + } + if candidate != inode { + continue + } + + addr = make([]byte, 4) + if _, err = hex.Decode(addr, []byte(match[2])); err != nil { + return + } + addr[0], addr[1], addr[2], addr[3] = addr[3], addr[2], addr[1], addr[0] + + // use a 32 bit int for target, at the result is a uint16 + port64, err = strconv.ParseInt(match[3], 16, 32) + if err != nil { + return + } + port = uint16(port64) + + return + } + + if err = scanner.Err(); err != nil { + return + } + + err = fmt.Errorf("Fd %d not found for proc %d", fd, pid) + return +} + +// in milliseconds +func now() int64 { + return time.Now().UnixNano() / 1000000 +} + +// We want to get the listening address from /proc +func newListeningFd(pid, fd int) (*Fd, error) { + localAddr, localPort, err := getLocalAddr(pid, fd) + if err != nil { + return nil, err + } + + return &Fd{ + fd: fd, + + ConnectionDetails: ConnectionDetails{ + direction: listening, + Start: now(), + ToAddr: localAddr, + ToPort: uint16(localPort), + }, + }, nil +} + +// We intercepted a connect syscall +func newConnectionFd(pid, fd int, remoteAddr net.IP, remotePort uint16) (*Fd, error) { + localAddr, localPort, err := getLocalAddr(pid, fd) + if err != nil { + return nil, err + } + + return &Fd{ + fd: fd, + + ConnectionDetails: ConnectionDetails{ + direction: outgoing, + Start: now(), + FromAddr: localAddr, + FromPort: uint16(localPort), + ToAddr: remoteAddr, + ToPort: remotePort, + }, + }, nil +} + +// We got a new connection on a listening socket +func (fd *Fd) newConnection(addr net.IP, port uint16, newFd int) (*Fd, error) { + if fd.direction != listening { + return nil, fmt.Errorf("New connection on non-listening fd!") + } + + return &Fd{ + fd: newFd, + + ConnectionDetails: ConnectionDetails{ + direction: incoming, + Start: now(), + ToAddr: fd.ToAddr, + ToPort: fd.ToPort, + FromAddr: addr, + FromPort: port, + }, + }, nil +} + +func (fd *Fd) close() { + fd.closed = true + fd.Stop = now() +} diff --git a/experimental/tracer/ptrace/process.go b/experimental/tracer/ptrace/process.go new file mode 100644 index 000000000..6add841c7 --- /dev/null +++ b/experimental/tracer/ptrace/process.go @@ -0,0 +1,99 @@ +package ptrace + +import ( + "fmt" + "io/ioutil" + "log" + "strconv" + "sync" +) + +type process struct { + sync.Mutex + pid int + + detaching bool + detached chan struct{} + + tracer *PTracer + threads map[int]*thread + fds map[int]*Fd +} + +func newProcess(pid int, tracer *PTracer) *process { + return &process{ + pid: pid, + tracer: tracer, + threads: make(map[int]*thread), + fds: make(map[int]*Fd), + detached: make(chan struct{}), + } +} + +func (p *process) trace() { + go p.loop() +} + +// This doesn't actually guarantees we follow all the threads. Oops. +func (p *process) loop() { + var ( + attached int + ) + log.Printf("Tracing process %d", p.pid) + + for { + ps, err := ioutil.ReadDir(fmt.Sprintf("/proc/%d/task", p.pid)) + if err != nil { + log.Printf("ReadDir failed, pid=%d, err=%v", p.pid, err) + return + } + + attached = 0 + for _, file := range ps { + pid, err := strconv.Atoi(file.Name()) + if err != nil { + log.Printf("'%s' is not a pid: %v", file.Name(), err) + attached++ + continue + } + + p.Lock() + t, ok := p.threads[pid] + if !ok { + t = p.tracer.traceThread(pid, p) + p.threads[pid] = t + } + p.Unlock() + + if !t.attached { + continue + } + + attached++ + } + + // When we successfully attach to all threads + // we can be sure to catch new clones, so we + // can quit. + if attached == len(ps) { + break + } + } + + log.Printf("Successfully attached to %d threads", attached) +} + +func (p *process) newThread(thread *thread) { + p.Lock() + defer p.Unlock() + p.threads[thread.tid] = thread +} + +func (p *process) newFd(fd *Fd) error { + _, ok := p.fds[fd.fd] + if ok { + return fmt.Errorf("New fd %d, alread exists!", fd.fd) + } + p.fds[fd.fd] = fd + return nil +} diff --git a/experimental/tracer/ptrace/ptracer.go b/experimental/tracer/ptrace/ptracer.go new file mode 100644 index 000000000..c0205b3f5 --- /dev/null +++ b/experimental/tracer/ptrace/ptracer.go @@ -0,0 +1,336 @@ +package ptrace + +import ( + "bufio" + "fmt" + "log" + "os" + "runtime" + "strconv" + "strings" + "syscall" +) + +const ( + ptraceOptions = syscall.PTRACE_O_TRACESYSGOOD | syscall.PTRACE_O_TRACECLONE + ptraceTracesysgoodBit = 0x80 +) + +// Store ... +type Store interface { + RecordConnection(int, *Fd) +} + +// PTracer ptrace processed and threads +type PTracer struct { + + // All ptrace calls must come from the + // same thread. So we wait on a separate + // thread. + ops chan func() + stopped chan stopped + quit chan struct{} + childAttached chan struct{} //used to signal the wait loop + + threads map[int]*thread + processes map[int]*process + store Store +} + +type stopped struct { + pid int + status syscall.WaitStatus +} + +// NewPTracer creates a new ptracer. +func NewPTracer(store Store) PTracer { + t := PTracer{ + ops: make(chan func()), + stopped: make(chan stopped), + quit: make(chan struct{}), + childAttached: make(chan struct{}), + + threads: make(map[int]*thread), + processes: make(map[int]*process), + store: store, + } + go t.waitLoop() + go t.loop() + return t +} + +// Stop stop stop +func (t *PTracer) Stop() { + out := make(chan []int) + t.ops <- func() { + pids := []int{} + for pid := range t.processes { + pids = append(pids, pid) + } + out <- pids + } + for _, pid := range <-out { + t.StopTracing(pid) + } + t.quit <- struct{}{} +} + +// TraceProcess starts tracing the given pid +func (t *PTracer) TraceProcess(pid int) *process { + result := make(chan *process) + t.ops <- func() { + process := newProcess(pid, t) + t.processes[pid] = process + process.trace() + result <- process + } + return <-result +} + +// StopTracing stops tracing all threads for the given pid +func (t *PTracer) StopTracing(pid int) error { + log.Printf("Detaching from %d", pid) + + errors := make(chan error) + processes := make(chan *process) + + t.ops <- func() { + // send sigstop to all threads + process, ok := t.processes[pid] + if !ok { + errors <- fmt.Errorf("PID %d not found", pid) + return + } + + // This flag tells the thread to detach when it next stops + process.detaching = true + + // Now send sigstop to all threads. + for _, thread := range process.threads { + log.Printf("sending SIGSTOP to %d", thread.tid) + if err := syscall.Tgkill(pid, thread.tid, syscall.SIGSTOP); err != nil { + errors <- err + return + } + } + + processes <- process + } + + select { + case err := <-errors: + return err + case process := <-processes: + <-process.detached + return nil + } +} + +// AttachedPIDs list the currently attached processes. +func (t *PTracer) AttachedPIDs() []int { + result := make(chan []int) + t.ops <- func() { + var pids []int + for pid := range t.processes { + pids = append(pids, pid) + } + result <- pids + } + return <-result +} + +func (t *PTracer) traceThread(pid int, process *process) *thread { + result := make(chan *thread) + t.ops <- func() { + thread := newThread(pid, process, t) + t.threads[pid] = thread + + err := syscall.PtraceAttach(pid) + if err != nil { + log.Printf("Attach %d failed: %v", pid, err) + return + } + + result <- thread + + select { + case t.childAttached <- struct{}{}: + default: + } + } + return <-result +} + +func (t *PTracer) waitLoop() { + var ( + status syscall.WaitStatus + pid int + err error + ) + + for { + //log.Printf("Waiting...") + pid, err = syscall.Wait4(-1, &status, syscall.WALL, nil) + if err != nil && err.(syscall.Errno) == syscall.ECHILD { + //log.Printf( "No children to wait4") + <-t.childAttached + continue + } + if err != nil { + log.Printf(" Wait failed: %v %d", err, err.(syscall.Errno)) + return + } + //log.Printf(" PID %d stopped with signal %#x", pid, status) + t.stopped <- stopped{pid, status} + } +} + +func (t *PTracer) loop() { + runtime.LockOSThread() + + for { + select { + case op := <-t.ops: + op() + case stopped := <-t.stopped: + t.handleStopped(stopped.pid, stopped.status) + case <-t.quit: + return + } + } +} + +func (t *PTracer) handleStopped(pid int, status syscall.WaitStatus) { + signal := syscall.Signal(0) + target, err := t.thread(pid) + if err != nil { + log.Printf("thread failed: %v", err) + return + } + + if !target.attached { + target.attached = true + + err = syscall.PtraceSetOptions(pid, ptraceOptions) + if err != nil { + log.Printf("SetOptions failed, pid=%d, err=%v", pid, err) + return + } + + } else if status.Stopped() && status.StopSignal() == syscall.SIGTRAP|ptraceTracesysgoodBit { + // pid entered Syscall-enter-stop or syscall-exit-stop + target.syscallStopped() + + } else if status.Stopped() && status.StopSignal() == syscall.SIGTRAP { + // pid entered PTRACE_EVENT stop + switch status.TrapCause() { + case syscall.PTRACE_EVENT_CLONE: + err := target.handleClone(pid) + if err != nil { + log.Printf("clone failed: %v", err) + return + } + default: + log.Printf("Unknown PTRACE_EVENT %d for pid %d", status.TrapCause(), pid) + } + } else if status.Exited() || status.Signaled() { + // "tracer can safely assume pid will exit" + t.threadExited(target) + return + } else if status.Stopped() { + // tracee recieved a non-trace related signal + signal = status.StopSignal() + + if signal == syscall.SIGSTOP && target.process.detaching { + t.detachThread(target) + return + } + } else { + // unknown stop - shouldn't happen! + log.Printf("Pid %d random stop with status %x", pid, status) + } + + // Restart stopped caller in syscall trap mode. + // log.Printf("Restarting pid %d with signal %d", pid, int(signal)) + err = syscall.PtraceSyscall(pid, int(signal)) + if err != nil { + log.Printf("PtraceSyscall failed, pid=%d, err=%v", pid, err) + } +} + +func (t *PTracer) detachThread(thread *thread) { + syscall.PtraceDetach(thread.tid) + process := thread.process + delete(process.threads, thread.tid) + delete(t.threads, thread.tid) + if len(process.threads) == 0 { + delete(t.processes, process.pid) + close(process.detached) + log.Printf("Process %d detached", process.pid) + } +} + +func pidForTid(tid int) (pid int, err error) { + var ( + status *os.File + scanner *bufio.Scanner + splits []string + ) + + status, err = os.Open(fmt.Sprintf("/proc/%d/status", tid)) + if err != nil { + return + } + defer status.Close() + + scanner = bufio.NewScanner(status) + for scanner.Scan() { + splits = strings.Split(scanner.Text(), ":") + if splits[0] != "Tgid" { + continue + } + + pid, err = strconv.Atoi(strings.TrimSpace(splits[1])) + return + } + + if err = scanner.Err(); err != nil { + return + } + + err = fmt.Errorf("Pid not found for proc %d", tid) + return +} + +func (t *PTracer) thread(tid int) (*thread, error) { + // unfortunately we can't propage fd affinitiy as we + // can reliably determin who clone'd new threads, as + // the pids returned by the ptrace calls are in the process' + // namespace, not ours. + + thread, ok := t.threads[tid] + if !ok { + pid, err := pidForTid(tid) + if err != nil { + return nil, err + } + + proc, ok := t.processes[pid] + if !ok { + return nil, fmt.Errorf("Got new thread %d for unknown process", tid) + } + + thread = newThread(tid, proc, t) + t.threads[tid] = thread + log.Printf("New thread reported, tid=%d, pid=%d", tid, pid) + } + return thread, nil +} + +func (t *PTracer) threadExited(thread *thread) { + thread.handleExit() + delete(t.threads, thread.tid) + if thread.process != nil { + delete(thread.process.threads, thread.tid) + } +} diff --git a/experimental/tracer/ptrace/thread.go b/experimental/tracer/ptrace/thread.go new file mode 100644 index 000000000..5a7b2dfad --- /dev/null +++ b/experimental/tracer/ptrace/thread.go @@ -0,0 +1,291 @@ +package ptrace + +import ( + "fmt" + "log" + "net" + "syscall" + "unsafe" +) + +// Syscall numbers +const ( + READ = 0 + WRITE = 1 + OPEN = 2 + CLOSE = 3 + STAT = 4 + MMAP = 9 + MPROTECT = 10 + MUNMAP = 11 + SELECT = 23 + MADVISE = 28 + SOCKET = 41 + CONNECT = 42 + ACCEPT = 43 + SENDTO = 44 + RECVFROM = 45 + SHUTDOWN = 48 + CLONE = 56 + GETTIMEOFDAY = 96 + GETID = 186 + FUTEX = 202 + SETROBUSTLIST = 273 + ACCEPT4 = 288 +) + +// States for a given thread +const ( + NORMAL = iota + INSYSCALL +) + +type thread struct { + tid int + attached bool + process *process // might be nil! + tracer *PTracer + + state int + callRegs syscall.PtraceRegs + resultRegs syscall.PtraceRegs + + currentIncoming map[int]*Fd + currentOutgoing map[int]*Fd + closedOutgoing []*Fd +} + +func newThread(pid int, process *process, tracer *PTracer) *thread { + t := &thread{ + tid: pid, + process: process, + tracer: tracer, + currentIncoming: map[int]*Fd{}, + currentOutgoing: map[int]*Fd{}, + } + return t +} + +// trace thread calls this +func (t *thread) syscallStopped() { + var err error + + if t.state == NORMAL { + if err = syscall.PtraceGetRegs(t.tid, &t.callRegs); err != nil { + t.logf("GetRegs failed, pid=%d, err=%v", t.tid, err) + } + t.state = INSYSCALL + return + } + + t.state = NORMAL + + if err = syscall.PtraceGetRegs(t.tid, &t.resultRegs); err != nil { + t.logf("GetRegs failed, pid=%d, err=%v", t.tid, err) + return + } + + if t.process == nil { + t.logf("Got syscall, but don't know parent process!") + return + } + + switch t.callRegs.Orig_rax { + case ACCEPT, ACCEPT4: + t.handleAccept(&t.callRegs, &t.resultRegs) + + case CLOSE: + t.handleClose(&t.callRegs, &t.resultRegs) + + case CONNECT: + t.handleConnect(&t.callRegs, &t.resultRegs) + + case READ, WRITE, RECVFROM, SENDTO: + t.handleIO(&t.callRegs, &t.resultRegs) + + // we can ignore these syscalls + case SETROBUSTLIST, GETID, MMAP, MPROTECT, MADVISE, SOCKET, CLONE, STAT, SELECT: + return + + case OPEN, FUTEX, SHUTDOWN, GETTIMEOFDAY, MUNMAP: + return + + default: + t.logf("syscall(%d)", t.callRegs.Orig_rax) + } +} + +func ntohl(b uint16) uint16 { + return (b << 8) | ((b & 0xff00) >> 8) +} + +func (t *thread) getSocketAddress(ptr uintptr) (addr net.IP, port uint16, err error) { + var ( + buf = make([]byte, syscall.SizeofSockaddrAny) + read int + ) + + if ptr == 0 { + err = fmt.Errorf("Null ptr") + return + } + + read, err = syscall.PtracePeekData(t.tid, ptr, buf) + if read != syscall.SizeofSockaddrAny || err != nil { + return + } + + var sockaddr4 = (*syscall.RawSockaddrInet4)(unsafe.Pointer(&buf[0])) + if sockaddr4.Family != syscall.AF_INET { + return + } + + addr = net.IP(sockaddr4.Addr[0:]) + port = ntohl(sockaddr4.Port) + return +} + +func (t *thread) handleAccept(call, result *syscall.PtraceRegs) { + var ( + err error + ok bool + listeningFdNum int + connectionFdNum int + addrPtr uintptr + addr net.IP + port uint16 + listeningFd *Fd + connection *Fd + ) + + listeningFdNum = int(result.Rdi) + connectionFdNum = int(result.Rax) + addrPtr = uintptr(result.Rsi) + addr, port, err = t.getSocketAddress(addrPtr) + if err != nil { + t.logf("failed to read sockaddr: %v", err) + return + } + + listeningFd, ok = t.process.fds[listeningFdNum] + if !ok { + listeningFd, err = newListeningFd(t.process.pid, listeningFdNum) + if err != nil { + t.logf("Failed to read listening port: %v", err) + return + } + t.process.fds[listeningFdNum] = listeningFd + } + + connection, err = listeningFd.newConnection(addr, port, connectionFdNum) + if err != nil { + t.logf("Failed to create connection fd: %v", err) + return + } + + t.process.newFd(connection) + + t.logf("Accepted connection from %s:%d -> %s:%d on fd %d, new fd %d", + addr, port, connection.ToAddr, connection.ToPort, listeningFdNum, connectionFdNum) +} + +func (t *thread) handleConnect(call, result *syscall.PtraceRegs) { + fd := int(result.Rdi) + ptr := result.Rsi + addr, port, err := t.getSocketAddress(uintptr(ptr)) + if err != nil { + t.logf("failed to read sockaddr: %v", err) + return + } + + connection, err := newConnectionFd(t.process.pid, fd, addr, port) + if err != nil { + t.logf("Failed to create connection fd: %v", err) + return + } + + t.process.newFd(connection) + + t.logf("Made connection from %s:%d -> %s:%d on fd %d", + connection.FromAddr, connection.FromPort, + connection.ToAddr, connection.ToPort, fd) +} + +func (t *thread) handleClose(call, result *syscall.PtraceRegs) { + fdNum := int(call.Rdi) + fd, ok := t.process.fds[fdNum] + if !ok { + t.logf("Got close unknown fd %d", fdNum) + return + } + + //t.logf("Closing fd %d", fdNum) + fd.close() + + // if this connection was incoming, add it to 'the registry' + if fd.direction == incoming { + // collect all the outgoing connections this thread has made + // and treat them as caused by this incoming connections + for _, outgoing := range t.currentOutgoing { + //t.logf("Fd %d caused %d", fdNum, outgoing.fd) + fd.Children = append(fd.Children, outgoing) + } + t.currentOutgoing = map[int]*Fd{} + + for _, outgoing := range t.closedOutgoing { + //t.logf("Fd %d caused %d", fdNum, outgoing.fd) + fd.Children = append(fd.Children, outgoing) + } + t.closedOutgoing = []*Fd{} + + t.tracer.store.RecordConnection(t.process.pid, fd) + } + + // now make sure we've remove it from everywhere + delete(t.process.fds, fdNum) + for _, thread := range t.process.threads { + delete(thread.currentIncoming, fdNum) + + if _, ok := thread.currentOutgoing[fdNum]; ok { + thread.closedOutgoing = append(thread.closedOutgoing, fd) + } + delete(thread.currentOutgoing, fdNum) + } +} + +func (t *thread) handleIO(call, result *syscall.PtraceRegs) { + fdNum := int(call.Rdi) + fd, ok := t.process.fds[fdNum] + if !ok { + //t.logf("IO on unknown fd %d", fdNum) + return + } + + if fd.direction == incoming { + //t.logf("IO on incoming connection %d; setting affinity", fdNum) + t.currentIncoming[fdNum] = fd + } else { + //t.logf("IO on outgoing connection %d; setting affinity", fdNum) + t.currentOutgoing[fdNum] = fd + } +} + +func (t *thread) handleClone(pid int) error { + // We can't use the pid in the process, as it may be in a namespace + newPid, err := syscall.PtraceGetEventMsg(pid) + if err != nil { + log.Printf("PtraceGetEventMsg failed: %v", err) + return err + } + + t.logf("New thread clone'd, pid=%d", newPid) + return nil +} + +func (t *thread) handleExit() { + t.logf("Exiting") +} + +func (t *thread) logf(fmt string, args ...interface{}) { + log.Printf("[thread %d] "+fmt, append([]interface{}{t.tid}, args...)...) +} diff --git a/experimental/tracer/tracer b/experimental/tracer/tracer new file mode 100755 index 000000000..3a35f0e5a --- /dev/null +++ b/experimental/tracer/tracer @@ -0,0 +1,47 @@ +#!/bin/bash + +set -eu + +usage() { + echo "$0" +} + +PORT=6060 +CONTAINER_NAME=weavetracer + +[ $# -gt 0 ] || usage +COMMAND=$1 +shift 1 + +case "$COMMAND" in + + launch) + docker run --privileged --net=host --pid=host -ti -v /var/run/docker.sock:/var/run/docker.sock \ + --name $CONTAINER_NAME weaveworks/tracer + ;; + + stop) + docker stop $CONTAINER_NAME || true + docker rm $CONTAINER_NAME >/dev/null || true + ;; + + attach) + PID=$1 + if [ -z "${PID##*[!0-9]*}" ]; then + PID=$(pgrep $PID) + fi + curl -X POST http://localhost:$PORT/pid/$PID + ;; + + detach) + PID=$1 + if [ -z "${PID##*[!0-9]*}" ]; then + PID=$(pgrep $PID) + fi + curl -X DELETE http://localhost:$PORT/pid/$PID + ;; + + traces) + curl http://localhost:$PORT/trace + ;; +esac diff --git a/experimental/tracer/ui/.gitignore b/experimental/tracer/ui/.gitignore new file mode 100644 index 000000000..a471811ad --- /dev/null +++ b/experimental/tracer/ui/.gitignore @@ -0,0 +1,4 @@ +container.json +traces.json +node_modules + diff --git a/experimental/tracer/ui/README.md b/experimental/tracer/ui/README.md new file mode 100644 index 000000000..a2dc141ef --- /dev/null +++ b/experimental/tracer/ui/README.md @@ -0,0 +1,10 @@ +Develop UI +########## + +* npm install +* npm start + +`server.js` serves a fake backend. You can add `container.json` and +`traces.json` to this directory to serve them as /traces and /container +respectively. + diff --git a/experimental/tracer/ui/favicon.ico b/experimental/tracer/ui/favicon.ico new file mode 100644 index 000000000..2d15c7808 Binary files /dev/null and b/experimental/tracer/ui/favicon.ico differ diff --git a/experimental/tracer/ui/index.html b/experimental/tracer/ui/index.html new file mode 100644 index 000000000..d7d92ef2c --- /dev/null +++ b/experimental/tracer/ui/index.html @@ -0,0 +1,393 @@ + + + + Weave Tracer + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/experimental/tracer/ui/logo.svg b/experimental/tracer/ui/logo.svg new file mode 100644 index 000000000..406732034 --- /dev/null +++ b/experimental/tracer/ui/logo.svg @@ -0,0 +1,54 @@ + + + + + + + + + + + + + + + + + + + + + \ No newline at end of file diff --git a/experimental/tracer/ui/package.json b/experimental/tracer/ui/package.json new file mode 100644 index 000000000..d2d91d3ed --- /dev/null +++ b/experimental/tracer/ui/package.json @@ -0,0 +1,15 @@ +{ + "name": "scope-tracer", + "version": "1.0.0", + "description": "", + "main": "server.js", + "scripts": { + "start": "node server.js", + "test": "echo \"Error: no test specified\" && exit 1" + }, + "author": "David Kaltschmidt ", + "license": "UNLICENSED", + "devDependencies": { + "express": "^4.13.3" + } +} diff --git a/experimental/tracer/ui/server.js b/experimental/tracer/ui/server.js new file mode 100644 index 000000000..e56d77342 --- /dev/null +++ b/experimental/tracer/ui/server.js @@ -0,0 +1,25 @@ +var express = require('express') + +var app = express(); + +app.get('/', function(req, res) { + res.sendFile(__dirname + '/index.html'); +}); + +app.get('/container', function(req, res) { + res.sendFile(__dirname + '/container.json'); +}); + +app.get('/traces', function(req, res) { + res.sendFile(__dirname + '/traces.json'); +}); + +app.use(express.static('./')); + +var port = process.env.PORT || 4050; +var server = app.listen(port, function () { + var host = server.address().address; + var port = server.address().port; + + console.log('Scope Tracer UI listening at http://%s:%s', host, port); +}); diff --git a/experimental/tracer/ui/sprintf.min.js b/experimental/tracer/ui/sprintf.min.js new file mode 100644 index 000000000..1cbc6b6eb --- /dev/null +++ b/experimental/tracer/ui/sprintf.min.js @@ -0,0 +1,4 @@ +/*! sprintf-js | Alexandru Marasteanu (http://alexei.ro/) | BSD-3-Clause */ + +!function(a){function b(){var a=arguments[0],c=b.cache;return c[a]&&c.hasOwnProperty(a)||(c[a]=b.parse(a)),b.format.call(null,c[a],arguments)}function c(a){return Object.prototype.toString.call(a).slice(8,-1).toLowerCase()}function d(a,b){return Array(b+1).join(a)}var e={not_string:/[^s]/,number:/[dief]/,json:/[j]/,not_json:/[^j]/,text:/^[^\x25]+/,modulo:/^\x25{2}/,placeholder:/^\x25(?:([1-9]\d*)\$|\(([^\)]+)\))?(\+)?(0|'[^$])?(-)?(\d+)?(?:\.(\d+))?([b-fijosuxX])/,key:/^([a-z_][a-z_\d]*)/i,key_access:/^\.([a-z_][a-z_\d]*)/i,index_access:/^\[(\d+)\]/,sign:/^[\+\-]/};b.format=function(a,f){var g,h,i,j,k,l,m,n=1,o=a.length,p="",q=[],r=!0,s="";for(h=0;o>h;h++)if(p=c(a[h]),"string"===p)q[q.length]=a[h];else if("array"===p){if(j=a[h],j[2])for(g=f[n],i=0;i=0),j[8]){case"b":g=g.toString(2);break;case"c":g=String.fromCharCode(g);break;case"d":case"i":g=parseInt(g,10);break;case"j":g=JSON.stringify(g,null,j[6]?parseInt(j[6]):0);break;case"e":g=j[7]?g.toExponential(j[7]):g.toExponential();break;case"f":g=j[7]?parseFloat(g).toFixed(j[7]):parseFloat(g);break;case"o":g=g.toString(8);break;case"s":g=(g=String(g))&&j[7]?g.substring(0,j[7]):g;break;case"u":g>>>=0;break;case"x":g=g.toString(16);break;case"X":g=g.toString(16).toUpperCase()}e.json.test(j[8])?q[q.length]=g:(!e.number.test(j[8])||r&&!j[3]?s="":(s=r?"+":"-",g=g.toString().replace(e.sign,"")),l=j[4]?"0"===j[4]?"0":j[4].charAt(1):" ",m=j[6]-(s+g).length,k=j[6]&&m>0?d(l,m):"",q[q.length]=j[5]?s+g+k:"0"===l?s+k+g:k+s+g)}return q.join("")},b.cache={},b.parse=function(a){for(var b=a,c=[],d=[],f=0;b;){if(null!==(c=e.text.exec(b)))d[d.length]=c[0];else if(null!==(c=e.modulo.exec(b)))d[d.length]="%";else{if(null===(c=e.placeholder.exec(b)))throw new SyntaxError("[sprintf] unexpected placeholder");if(c[2]){f|=1;var g=[],h=c[2],i=[];if(null===(i=e.key.exec(h)))throw new SyntaxError("[sprintf] failed to parse named argument key");for(g[g.length]=i[1];""!==(h=h.substring(i[0].length));)if(null!==(i=e.key_access.exec(h)))g[g.length]=i[1];else{if(null===(i=e.index_access.exec(h)))throw new SyntaxError("[sprintf] failed to parse named argument key");g[g.length]=i[1]}c[2]=g}else f|=2;if(3===f)throw new Error("[sprintf] mixing positional and named placeholders is not (yet) supported");d[d.length]=c}b=b.substring(c[0].length)}return d};var f=function(a,c,d){return d=(c||[]).slice(0),d.splice(0,0,a),b.apply(null,d)};"undefined"!=typeof exports?(exports.sprintf=b,exports.vsprintf=f):(a.sprintf=b,a.vsprintf=f,"function"==typeof define&&define.amd&&define(function(){return{sprintf:b,vsprintf:f}}))}("undefined"==typeof window?this:window); +//# sourceMappingURL=sprintf.min.js.map \ No newline at end of file diff --git a/probe/docker/container.go b/probe/docker/container.go index 5823e3739..b2553317b 100644 --- a/probe/docker/container.go +++ b/probe/docker/container.go @@ -86,7 +86,7 @@ type Container interface { GetNode(string, []net.IP) report.Node State() string HasTTY() bool - + Container() *docker.Container StartGatheringStats() error StopGatheringStats() } @@ -146,6 +146,10 @@ func (c *container) State() string { return StateStopped } +func (c *container) Container() *docker.Container { + return c.container +} + func (c *container) StartGatheringStats() error { c.Lock() defer c.Unlock() diff --git a/probe/docker/registry_test.go b/probe/docker/registry_test.go index caba06996..73f175455 100644 --- a/probe/docker/registry_test.go +++ b/probe/docker/registry_test.go @@ -56,6 +56,10 @@ func (c *mockContainer) GetNode(_ string, _ []net.IP) report.Node { }) } +func (c *mockContainer) Container() *client.Container { + return c.c +} + func (c *mockContainer) HasTTY() bool { return true } type mockDockerClient struct { diff --git a/probe/docker/tagger_test.go b/probe/docker/tagger_test.go index f1811f6d4..1415d4e1b 100644 --- a/probe/docker/tagger_test.go +++ b/probe/docker/tagger_test.go @@ -23,6 +23,10 @@ func (m *mockProcessTree) GetParent(pid int) (int, error) { return parent, nil } +func (m *mockProcessTree) GetChildren(int) ([]int, error) { + panic("Not implemented") +} + func TestTagger(t *testing.T) { oldProcessTree := docker.NewProcessTreeStub defer func() { docker.NewProcessTreeStub = oldProcessTree }() diff --git a/probe/process/tree.go b/probe/process/tree.go index 27f379009..8c3bde7c3 100644 --- a/probe/process/tree.go +++ b/probe/process/tree.go @@ -7,6 +7,7 @@ import ( // Tree represents all processes on the machine. type Tree interface { GetParent(pid int) (int, error) + GetChildren(pid int) ([]int, error) } type tree struct { @@ -32,3 +33,30 @@ func (pt *tree) GetParent(pid int) (int, error) { return proc.PPID, nil } + +// GetChildren +func (pt *tree) GetChildren(pid int) ([]int, error) { + _, ok := pt.processes[pid] + if !ok { + return []int{}, fmt.Errorf("PID %d not found", pid) + } + + var isChild func(id int) bool + isChild = func(id int) bool { + p, ok := pt.processes[id] + if !ok || p.PPID == 0 { + return false + } else if p.PPID == pid { + return true + } + return isChild(p.PPID) + } + + children := []int{pid} + for id := range pt.processes { + if isChild(id) { + children = append(children, id) + } + } + return children, nil +}