Skip to content

Commit 2f09e6a

Browse files
kcqkcq
committed
ipc stability enhancements and cleanup
1 parent 4167f6a commit 2f09e6a

File tree

9 files changed

+112
-51
lines changed

9 files changed

+112
-51
lines changed

‎internal/app/master/inspectors/container/container_inspector.go‎

Lines changed: 10 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -14,7 +14,7 @@ import (
1414
"github.com/docker-slim/docker-slim/internal/app/master/inspectors/image"
1515
"github.com/docker-slim/docker-slim/internal/app/master/security/apparmor"
1616
"github.com/docker-slim/docker-slim/internal/app/master/security/seccomp"
17-
"github.com/docker-slim/docker-slim/pkg/messages"
17+
"github.com/docker-slim/docker-slim/pkg/ipc/command"
1818
"github.com/docker-slim/docker-slim/pkg/report"
1919
"github.com/docker-slim/docker-slim/pkg/utils/errutils"
2020
"github.com/docker-slim/docker-slim/pkg/utils/fsutils"
@@ -177,7 +177,7 @@ func (i *Inspector) RunContainer() error {
177177
return err
178178
}
179179

180-
cmd := &messages.StartMonitor{
180+
cmd := &command.StartMonitor{
181181
AppName: i.FatContainerCmd[0],
182182
}
183183

@@ -259,16 +259,17 @@ func (i *Inspector) ShutdownContainer() error {
259259

260260
// FinishMonitoring ends the target container monitoring activities
261261
func (i *Inspector) FinishMonitoring() {
262-
cmdResponse, err := ipc.SendContainerCmd(&messages.StopMonitor{})
262+
cmdResponse, err := ipc.SendContainerCmd(&command.StopMonitor{})
263263
errutils.WarnOn(err)
264-
_ = cmdResponse
264+
//_ = cmdResponse
265+
log.Debugf("'stop' monitor response => '%v'", cmdResponse)
265266

266-
log.Debugf("'stop' response => '%v'", cmdResponse)
267267
log.Info("docker-slim: waiting for the container to finish its work...")
268268

269269
//for now there's only one event ("done")
270270
//getEvt() should timeout in two minutes (todo: pick a good timeout)
271271
evt, err := ipc.GetContainerEvt()
272+
log.Debugf("sensor event => '%v'", evt)
272273

273274
//don't want to expose mangos here... mangos.ErrRecvTimeout = errors.New("receive time out")
274275
if err != nil && err.Error() == IpcErrRecvTimeoutStr {
@@ -279,6 +280,10 @@ func (i *Inspector) FinishMonitoring() {
279280
errutils.WarnOn(err)
280281
_ = evt
281282
log.Debugf("docker-slim: sensor event => '%v'", evt)
283+
284+
cmdResponse, err = ipc.SendContainerCmd(&command.ShutdownSensor{})
285+
errutils.WarnOn(err)
286+
log.Debugf("'shutdown' sensor response => '%v'", cmdResponse)
282287
}
283288

284289
func (i *Inspector) initContainerChannels() error {

‎internal/app/master/inspectors/container/ipc/ipc.go‎

Lines changed: 13 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -11,7 +11,9 @@ import (
1111
//"github.com/go-mangos/mangos/transport/ipc"
1212
"github.com/go-mangos/mangos/transport/tcp"
1313

14-
"github.com/docker-slim/docker-slim/pkg/messages"
14+
"github.com/docker-slim/docker-slim/pkg/ipc/channel"
15+
"github.com/docker-slim/docker-slim/pkg/ipc/command"
16+
"github.com/docker-slim/docker-slim/pkg/ipc/event"
1517
)
1618

1719
// InitContainerChannels initializes the communication channels with the target container
@@ -37,12 +39,12 @@ func InitContainerChannels(dockerHostIP, cmdChannelPort, evtChannelPort string)
3739
}
3840

3941
// SendContainerCmd sends the given command to the target container
40-
func SendContainerCmd(cmd messages.Message) (string, error) {
42+
func SendContainerCmd(cmd command.Message) (string, error) {
4143
return sendCmd(cmdChannel, cmd)
4244
}
4345

4446
// GetContainerEvt returns the current event generated by the target container
45-
func GetContainerEvt() (string, error) {
47+
func GetContainerEvt() (event.Name, error) {
4648
return getEvt(evtChannel)
4749
}
4850

@@ -53,7 +55,8 @@ func ShutdownContainerChannels() {
5355
}
5456

5557
//var cmdChannelAddr = "ipc:///tmp/docker-slim-sensor.cmds.ipc"
56-
var cmdChannelAddr = "tcp://127.0.0.1:65501"
58+
var cmdChannelAddr = fmt.Sprintf("tcp://127.0.0.1:%d", channel.CmdPort)
59+
5760
var cmdChannel mangos.Socket
5861

5962
func newCmdClient(addr string) (mangos.Socket, error) {
@@ -89,13 +92,13 @@ func shutdownCmdChannel() {
8992
}
9093
}
9194

92-
func sendCmd(channel mangos.Socket, cmd messages.Message) (string, error) {
95+
func sendCmd(channel mangos.Socket, cmd command.Message) (string, error) {
9396
sendTimeouts := 0
9497
recvTimeouts := 0
9598

9699
log.Debugf("sendCmd(%s)", cmd)
97100
for {
98-
sendData, err := messages.Encode(cmd)
101+
sendData, err := command.Encode(cmd)
99102
if err != nil {
100103
log.Info("sendCmd(): malformed cmd - ", err)
101104
return "", err
@@ -132,7 +135,7 @@ func sendCmd(channel mangos.Socket, cmd messages.Message) (string, error) {
132135
}
133136
}
134137

135-
var evtChannelAddr = "tcp://127.0.0.1:65502"
138+
var evtChannelAddr = fmt.Sprintf("tcp://127.0.0.1:%d", channel.EvtPort)
136139

137140
//var evtChannelAddr = "ipc:///tmp/docker-slim-sensor.events.ipc"
138141
var evtChannel mangos.Socket
@@ -170,12 +173,13 @@ func shutdownEvtChannel() {
170173
}
171174
}
172175

173-
func getEvt(channel mangos.Socket) (string, error) {
176+
func getEvt(channel mangos.Socket) (event.Name, error) {
174177
log.Debug("getEvt()")
175178
evt, err := channel.Recv()
179+
log.Debug("getEvt(): channel.Recv() - done")
176180
if err != nil {
177181
return "", err
178182
}
179183

180-
return string(evt), nil
184+
return event.Name(evt), nil
181185
}

‎internal/app/sensor/app.go‎

Lines changed: 19 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -9,7 +9,8 @@ import (
99
"github.com/docker-slim/docker-slim/internal/app/sensor/monitors/fanotify"
1010
"github.com/docker-slim/docker-slim/internal/app/sensor/monitors/pevent"
1111
"github.com/docker-slim/docker-slim/internal/app/sensor/monitors/ptrace"
12-
"github.com/docker-slim/docker-slim/pkg/messages"
12+
"github.com/docker-slim/docker-slim/pkg/ipc/command"
13+
"github.com/docker-slim/docker-slim/pkg/ipc/event"
1314
"github.com/docker-slim/docker-slim/pkg/report"
1415
"github.com/docker-slim/docker-slim/pkg/utils/errutils"
1516

@@ -25,7 +26,7 @@ func monitor(stopWork chan bool,
2526
stopWorkAck chan bool,
2627
pids chan []int,
2728
ptmonStartChan chan int,
28-
cmd *messages.StartMonitor,
29+
cmd *command.StartMonitor,
2930
dirName string) {
3031
log.Info("sensor: monitor starting...")
3132
mountPoint := "/"
@@ -119,22 +120,32 @@ doneRunning:
119120
case cmd := <-cmdChan:
120121
log.Debug("\nsensor: command => ", cmd)
121122
switch data := cmd.(type) {
122-
case *messages.StartMonitor:
123+
case *command.StartMonitor:
123124
if data == nil {
124-
log.Info("sensor: 'start' command - no data...")
125+
log.Info("sensor: 'start' monitor command - no data...")
125126
break
126127
}
127128

128-
log.Debugf("sensor: 'start' command (%#v) - starting monitor...", data)
129+
log.Debugf("sensor: 'start' monitor command (%#v)", data)
129130
monitor(monDoneChan, monDoneAckChan, pidsChan, ptmonStartChan, data, dirName)
130131

131132
//target app started by ptmon... (long story :-))
132133
//TODO: need to get the target app pid to pemon, so it can filter process events
133134
log.Debugf("sensor: target app started => %v %#v", data.AppName, data.AppArgs)
134135
time.Sleep(3 * time.Second)
135136

136-
case *messages.StopMonitor:
137-
log.Debug("sensor: 'stop' command - stopping monitor...")
137+
case *command.StopMonitor:
138+
log.Debug("sensor: 'stop' monitor command")
139+
140+
monDoneChan <- true
141+
log.Info("sensor: waiting for monitor to finish...")
142+
<-monDoneAckChan
143+
log.Info("sensor: monitor stopped...")
144+
145+
ipc.TryPublishEvt(3, event.StopMonitorDoneName)
146+
147+
case *command.ShutdownSensor:
148+
log.Debug("sensor: 'shutdown' sensor command")
138149
break doneRunning
139150
default:
140151
log.Debug("sensor: ignoring unknown command => ", cmd)
@@ -145,11 +156,7 @@ doneRunning:
145156
}
146157
}
147158

148-
monDoneChan <- true
149-
log.Info("sensor: waiting for monitor to finish...")
150-
<-monDoneAckChan
151-
152-
ipc.TryPublishEvt(3, "monitor.finish.completed")
159+
ipc.TryPublishEvt(3, event.ShutdownSensorDoneName)
153160

154161
log.Info("sensor: done!")
155162
}

‎internal/app/sensor/artifacts.go‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,7 @@ import (
1616
"strings"
1717
"syscall"
1818

19-
"github.com/docker-slim/docker-slim/pkg/messages"
19+
"github.com/docker-slim/docker-slim/pkg/ipc/command"
2020
"github.com/docker-slim/docker-slim/pkg/report"
2121
"github.com/docker-slim/docker-slim/pkg/utils/errutils"
2222
"github.com/docker-slim/docker-slim/pkg/utils/fsutils"
@@ -54,7 +54,7 @@ func saveResults(fanMonReport *report.FanMonitorReport,
5454
fileNames map[string]*report.ArtifactProps,
5555
ptMonReport *report.PtMonitorReport,
5656
peReport *report.PeMonitorReport,
57-
cmd *messages.StartMonitor) {
57+
cmd *command.StartMonitor) {
5858
artifactDirName := defaultArtifactDirName
5959

6060
artifactStore := newArtifactStore(artifactDirName, fanMonReport, fileNames, ptMonReport, peReport, cmd)
@@ -73,15 +73,15 @@ type artifactStore struct {
7373
resolve map[string]struct{}
7474
linkMap map[string]*report.ArtifactProps
7575
fileMap map[string]*report.ArtifactProps
76-
cmd *messages.StartMonitor
76+
cmd *command.StartMonitor
7777
}
7878

7979
func newArtifactStore(storeLocation string,
8080
fanMonReport *report.FanMonitorReport,
8181
rawNames map[string]*report.ArtifactProps,
8282
ptMonReport *report.PtMonitorReport,
8383
peMonReport *report.PeMonitorReport,
84-
cmd *messages.StartMonitor) *artifactStore {
84+
cmd *command.StartMonitor) *artifactStore {
8585
store := &artifactStore{
8686
storeLocation: storeLocation,
8787
fanMonReport: fanMonReport,

‎internal/app/sensor/data_porcessor.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ import (
1010
"syscall"
1111

1212
"github.com/docker-slim/docker-slim/internal/app/sensor/monitors/fanotify"
13-
"github.com/docker-slim/docker-slim/pkg/messages"
13+
"github.com/docker-slim/docker-slim/pkg/ipc/command"
1414
"github.com/docker-slim/docker-slim/pkg/report"
1515

1616
log "github.com/Sirupsen/logrus"
@@ -20,7 +20,7 @@ func processReports(mountPoint string,
2020
fanReport *report.FanMonitorReport,
2121
ptReport *report.PtMonitorReport,
2222
peReport *report.PeMonitorReport,
23-
cmd *messages.StartMonitor) {
23+
cmd *command.StartMonitor) {
2424

2525
fileCount := 0
2626
for _, processFileMap := range fanReport.ProcessFiles {

‎internal/app/sensor/ipc/ipc.go‎

Lines changed: 21 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
package ipc
22

33
import (
4+
"fmt"
45
"time"
56

67
log "github.com/Sirupsen/logrus"
@@ -10,7 +11,9 @@ import (
1011
//"github.com/go-mangos/mangos/transport/ipc"
1112
"github.com/go-mangos/mangos/transport/tcp"
1213

13-
"github.com/docker-slim/docker-slim/pkg/messages"
14+
"github.com/docker-slim/docker-slim/pkg/ipc/channel"
15+
"github.com/docker-slim/docker-slim/pkg/ipc/command"
16+
"github.com/docker-slim/docker-slim/pkg/ipc/event"
1417
)
1518

1619
// InitChannels initializes the communication channels with the master
@@ -36,11 +39,11 @@ func ShutdownChannels() {
3639
}
3740

3841
// RunCmdServer starts the command server
39-
func RunCmdServer(done <-chan struct{}) (<-chan messages.Message, error) {
42+
func RunCmdServer(done <-chan struct{}) (<-chan command.Message, error) {
4043
return runCmdServer(cmdChannel, done)
4144
}
4245

43-
var cmdChannelAddr = "tcp://0.0.0.0:65501"
46+
var cmdChannelAddr = fmt.Sprintf("tcp://0.0.0.0:%d", channel.CmdPort)
4447

4548
//var cmdChannelAddr = "ipc:///tmp/docker-slim-sensor.cmds.ipc"
4649
//var cmdChannelAddr = "ipc:///opt/dockerslim/ipc/docker-slim-sensor.cmds.ipc"
@@ -68,8 +71,8 @@ func newCmdServer(addr string) (mangos.Socket, error) {
6871
return socket, nil
6972
}
7073

71-
func runCmdServer(channel mangos.Socket, done <-chan struct{}) (<-chan messages.Message, error) {
72-
cmdChan := make(chan messages.Message)
74+
func runCmdServer(channel mangos.Socket, done <-chan struct{}) (<-chan command.Message, error) {
75+
cmdChan := make(chan command.Message)
7376
go func() {
7477
for {
7578
// Could also use sock.RecvMsg to get header
@@ -89,7 +92,7 @@ func runCmdServer(channel mangos.Socket, done <-chan struct{}) (<-chan messages.
8992
} else {
9093
log.Debug("sensor: cmd server - got a command => ", string(rawCmd))
9194

92-
if cmd, err := messages.Decode(rawCmd); err != nil {
95+
if cmd, err := command.Decode(rawCmd); err != nil {
9396
log.Println(err)
9497
} else {
9598
cmdChan <- cmd
@@ -99,10 +102,10 @@ func runCmdServer(channel mangos.Socket, done <-chan struct{}) (<-chan messages.
99102
//NOTE:
100103
//must reply before receiving the next message
101104
//otherwise nanomsg/mangos will be confused :-)
102-
monitorFinishReply := "ok"
103-
err = channel.Send([]byte(monitorFinishReply))
105+
cmdStatusReply := "ok"
106+
err = channel.Send([]byte(cmdStatusReply))
104107
if err != nil {
105-
log.Warnln("sensor: cmd server - fail to send monitor.finish reply =>", err)
108+
log.Warnln("sensor: cmd server - fail to send command status reply =>", err)
106109
}
107110
}
108111
}
@@ -119,7 +122,7 @@ func shutdownCmdChannel() {
119122
}
120123
}
121124

122-
var evtChannelAddr = "tcp://0.0.0.0:65502"
125+
var evtChannelAddr = fmt.Sprintf("tcp://0.0.0.0:%d", channel.EvtPort)
123126

124127
//var evtChannelAddr = "ipc:///tmp/docker-slim-sensor.events.ipc"
125128
//var evtChannelAddr = "ipc:///opt/dockerslim/ipc/docker-slim-sensor.events.ipc"
@@ -147,20 +150,23 @@ func newEvtPublisher(addr string) (mangos.Socket, error) {
147150
return socket, nil
148151
}
149152

150-
func publishEvt(channel mangos.Socket, evt string) error {
151-
if err := channel.Send([]byte(evt)); err != nil {
152-
log.Debugf("fail to publish '%v' event:%v", evt, err)
153+
func publishEvt(channel mangos.Socket, event event.Name) error {
154+
log.Debugf("publishEvt(%v)", event)
155+
if err := channel.Send([]byte(event)); err != nil {
156+
log.Debugf("fail to publish '%v' event:%v", event, err)
153157
return err
154158
}
155159

156160
return nil
157161
}
158162

159163
// TryPublishEvt attempts to publish an event to the master
160-
func TryPublishEvt(ptry uint, event string) {
164+
func TryPublishEvt(ptry uint, event event.Name) {
165+
log.Debugf("TryPublishEvt(%v,%v)", ptry, event)
166+
161167
for ptry := 0; ptry < 3; ptry++ {
162168
log.Debugf("sensor: trying to publish '%v' event (attempt %v)", event, ptry+1)
163-
err := publishEvt(evtChannel, "monitor.finish.completed")
169+
err := publishEvt(evtChannel, event)
164170
if err == nil {
165171
log.Infof("sensor: published '%v'", event)
166172
break

‎pkg/ipc/channel/channel.go‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,7 @@
1+
package channel
2+
3+
// Supported events
4+
const (
5+
CmdPort = 65501
6+
EvtPort = 65502
7+
)

0 commit comments

Comments
 (0)