Merge pull request #752 from awprice/issue-751

add grpc endpoint to retrieve oom events

agent.go

@@ -144,6 +144,7 @@ type sandbox struct {
 	sandboxPidNs      bool
 	storages          map[string]*sandboxStorage
 	stopServer        chan struct{}
+	oomEvents         chan string
 }
 
 var agentFields = logrus.Fields{
@@ -814,6 +815,19 @@ func (s *sandbox) listenToUdevEvents() {
 	}
 }
 
+func (s *sandbox) runOOMEventMonitor(ch <-chan struct{}, containerID string) {
+	go func() {
+		for {
+			_, ok := <-ch
+			if !ok {
+				return
+			}
+			agentLog.WithField("container-id", containerID).Info("Received OOM event")
+			s.oomEvents <- containerID
+		}
+	}()
+}
+
 // This loop is meant to be run inside a separate Go routine.
 func (s *sandbox) signalHandlerLoop(sigCh chan os.Signal, errCh chan error) {
 	// Lock OS thread as subreaper is a thread local capability
@@ -1537,6 +1551,7 @@ func realMain() error {
 		deviceWatchers: make(map[string](chan string)),
 		storages:       make(map[string]*sandboxStorage),
 		stopServer:     make(chan struct{}),
+		oomEvents:      make(chan string),
 	}
 
 	rootSpan, rootContext, err = setupTracing(agentName)

agent_test.go

@@ -971,3 +971,21 @@ func TestSetupDebugConsole(t *testing.T) {
 		assert.NoError(err, msg)
 	}
 }
+
+func TestRunOOMEventMonitor(t *testing.T) {
+	assert := assert.New(t)
+
+	cid := "foo"
+	eventChan := make(chan struct{})
+	s := &sandbox{
+		oomEvents: make(chan string),
+	}
+
+	s.runOOMEventMonitor(eventChan, cid)
+
+	eventChan <- struct{}{}
+	oomEvent := <-s.oomEvents
+	assert.Equal(cid, oomEvent)
+
+	close(eventChan)
+}

grpc.go

@@ -861,6 +861,13 @@ func (a *agentGRPC) StartContainer(ctx context.Context, req *pb.StartContainerRe
 		return emptyResp, err
 	}
 
+	// Add the container to the OOM event monitor
+	oomCh, err := ctr.container.NotifyOOM()
+	if err != nil {
+		return emptyResp, err
+	}
+	a.sandbox.runOOMEventMonitor(oomCh, req.ContainerId)
+
 	return emptyResp, nil
 }
 
@@ -1792,6 +1799,11 @@ func (a *agentGRPC) StopTracing(ctx context.Context, req *pb.StopTracingRequest)
 	return emptyResp, nil
 }
 
+func (a *agentGRPC) GetOOMEvent(ctx context.Context, req *pb.GetOOMEventRequest) (*pb.OOMEvent, error) {
+	containerID := <-a.sandbox.oomEvents
+	return &pb.OOMEvent{ContainerId: containerID}, nil
+}
+
 // createExtendedPipe creates a pipe.
 // Optionally extends the pipe if containerPipeSize is positive.
 func createExtendedPipe() (*os.File, *os.File, error) {

grpc_test.go

@@ -1821,6 +1821,53 @@ func TestCreateExtendedPipe(t *testing.T) {
 	assert.Equal(containerPipeSize, size)
 }
 
+func TestGetOOMEvent(t *testing.T) {
+	assert := assert.New(t)
+
+	a := &agentGRPC{
+		sandbox: &sandbox{
+			oomEvents: make(chan string),
+		},
+	}
+
+	cid1 := "foo"
+	cid2 := "bar"
+
+	cid1EventChan := make(chan struct{})
+	cid2EventChan := make(chan struct{})
+
+	// Start the oom monitors
+	a.sandbox.runOOMEventMonitor(cid1EventChan, cid1)
+	a.sandbox.runOOMEventMonitor(cid2EventChan, cid2)
+
+	// Test sending two oom events
+	cid1EventChan <- struct{}{}
+
+	req := &pb.GetOOMEventRequest{}
+
+	// Retrieve both events
+	oomEventRes, err := a.GetOOMEvent(context.Background(), req)
+	assert.NoError(err)
+	assert.Equal(cid1, oomEventRes.ContainerId)
+
+	cid2EventChan <- struct{}{}
+
+	oomEventRes, err = a.GetOOMEvent(context.Background(), req)
+	assert.NoError(err)
+	assert.Equal(cid2, oomEventRes.ContainerId)
+
+	close(cid1EventChan)
+
+	cid2EventChan <- struct{}{}
+
+	// Ensure that cid2 still works
+	oomEventRes, err = a.GetOOMEvent(context.Background(), req)
+	assert.NoError(err)
+	assert.Equal(cid2, oomEventRes.ContainerId)
+
+	close(cid2EventChan)
+}
+
 func getPipeSize(f *os.File) (uint32, error) {
 	r1, _, err := syscall.Syscall(syscall.SYS_FCNTL, f.Fd(), syscall.F_GETPIPE_SZ, 0)
 	return uint32(r1), err