added handler functions, todo: write test

Author: carolinaecalderon

master/internal/rm/kubernetesrm/informer.go

@@ -17,27 +17,34 @@ import (
 
 const defaultBackoff = 5 * time.Second
 
+type podCallbackFunc func(*k8sV1.Pod)
+
 type informer struct {
 	syslog       *logrus.Entry
 	podInterface typedV1.PodInterface
+	podHandler   podCallbackFunc
 }
 
 func newInformer(
 	namespace string,
 	podInterface typedV1.PodInterface,
+	podHandler func(*k8sV1.Pod),
 ) *informer {
 	return &informer{
 		syslog:       logrus.WithField("Informer", namespace),
 		podInterface: podInterface,
+		podHandler:   podHandler,
 	}
 }
 
 // startInformer returns the updated pod, if any.
-func (i *informer) startInformer() (*k8sV1.Pod, bool) {
+func (i *informer) startInformer() bool {
+	ok := true
 	pods, err := i.podInterface.List(
 		context.TODO(), metaV1.ListOptions{LabelSelector: determinedLabel})
 	if err != nil {
 		i.syslog.WithError(err).Warnf("error retrieving internal resource version")
+		ok = false
 	}
 
 	rw, err := watchtools.NewRetryWatcher(pods.ResourceVersion, &cache.ListWatch{
@@ -48,6 +55,7 @@ func (i *informer) startInformer() (*k8sV1.Pod, bool) {
 	})
 	if err != nil {
 		i.syslog.WithError(err).Warnf("error initializing pod retry watcher")
+		ok = false
 	}
 
 	i.syslog.Info("pod informer is starting")
@@ -57,16 +65,16 @@ func (i *informer) startInformer() (*k8sV1.Pod, bool) {
 			continue
 		}
 
-		pod, ok := event.Object.(*k8sV1.Pod)
-		if !ok {
+		pod, OK := event.Object.(*k8sV1.Pod)
+		if !OK { // TODO CAROLINA: capitalizing OK to differentiate from ok var, better name?
 			i.syslog.Warnf("error converting event of type %T to *k8sV1.Pod: %+v", event, event)
 			continue
 		}
 
 		i.syslog.Debugf("informer got new pod event for pod: %s %s", pod.Name, pod.Status.Phase)
-		return pod, true
+		go i.podHandler(pod)
 	}
 
 	i.syslog.Warn("pod informer stopped unexpectedly")
-	return nil, false
+	return ok
 }

master/internal/rm/kubernetesrm/informer_intg_test.go

@@ -0,0 +1,25 @@
+package kubernetesrm
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	k8sV1 "k8s.io/api/core/v1"
+)
+
+// Testing informer.go.
+func mockPodHandler(pod *k8sV1.Pod) {}
+
+func TestStartInformer(t *testing.T) {
+	podInterface := &mockPodInterface{}
+	mockLogMessage := "mock log message"
+	podInterface = &mockPodInterface{logMessage: &mockLogMessage}
+	i := newInformer("testing", podInterface, mockPodHandler)
+	assert.NotNil(t, i)
+}
+
+// Testing nodes.go.
+
+// TODO Testing events.go.
+
+// TODO Testing preemptions.go.

master/internal/rm/kubernetesrm/nodes.go

@@ -9,37 +9,34 @@ import (
 	"k8s.io/client-go/tools/cache"
 )
 
-// Messages sent by the nodeInformer to the podsHandler.
-type nodeStatusUpdate struct {
-	updatedNode *k8sV1.Node
-	deletedNode *k8sV1.Node
-}
+type nodeCallbackFunc func(*k8sV1.Node, bool)
 
 type nodeInformer struct {
-	informer k8Informers.SharedInformerFactory
-	syslog   *logrus.Entry
-	stop     chan struct{}
+	informer    k8Informers.SharedInformerFactory
+	nodeHandler nodeCallbackFunc
+	syslog      *logrus.Entry
+	stop        chan struct{}
 }
 
-func newNodeInformer(clientSet k8sClient.Interface) *nodeInformer {
+func newNodeInformer(clientSet k8sClient.Interface, nodeHandler nodeCallbackFunc) *nodeInformer {
 	return &nodeInformer{
 		informer: k8Informers.NewSharedInformerFactoryWithOptions(
 			clientSet, 0, []k8Informers.SharedInformerOption{}...),
-		syslog: logrus.WithField("component", "nodeInformer"),
-		stop:   make(chan struct{}),
+		nodeHandler: nodeHandler,
+		syslog:      logrus.WithField("component", "nodeInformer"),
+		stop:        make(chan struct{}),
 	}
 }
 
-func (n *nodeInformer) startNodeInformer() (nodeStatusUpdate, bool) {
-	nodeStatus := nodeStatusUpdate{}
+func (n *nodeInformer) startNodeInformer() bool {
 	err := true
 	nodeInformer := n.informer.Core().V1().Nodes().Informer()
 	nodeInformer.AddEventHandler(&cache.ResourceEventHandlerFuncs{
 		AddFunc: func(obj interface{}) {
 			node, ok := obj.(*k8sV1.Node)
 			if ok {
 				n.syslog.Debugf("node added %s", node.Name)
-				nodeStatus.updatedNode = node
+				go n.nodeHandler(node, true)
 			} else {
 				n.syslog.Warnf("error converting event of type %T to *k8sV1.Node: %+v", obj, obj)
 				err = false
@@ -49,7 +46,7 @@ func (n *nodeInformer) startNodeInformer() (nodeStatusUpdate, bool) {
 			node, ok := newObj.(*k8sV1.Node)
 			if ok {
 				n.syslog.Debugf("node updated %s", node.Name)
-				nodeStatus.updatedNode = node
+				go n.nodeHandler(node, true)
 			} else {
 				n.syslog.Warnf("error converting event of type %T to *k8sV1.Node: %+v", newObj, newObj)
 				err = false
@@ -59,7 +56,7 @@ func (n *nodeInformer) startNodeInformer() (nodeStatusUpdate, bool) {
 			node, ok := obj.(*k8sV1.Node)
 			if ok {
 				n.syslog.Debugf("node stopped %s", node.Name)
-				nodeStatus.deletedNode = node
+				go n.nodeHandler(node, false)
 			} else {
 				n.syslog.Warnf("error converting event of type %T to *k8sV1.Node: %+v", obj, obj)
 				err = false
@@ -73,7 +70,7 @@ func (n *nodeInformer) startNodeInformer() (nodeStatusUpdate, bool) {
 	}
 	n.syslog.Info("node informer has started")
 
-	return nodeStatus, err
+	return err
 }
 
 func (n *nodeInformer) stopNodeInformer() {

master/internal/rm/kubernetesrm/pods.go

@@ -243,7 +243,7 @@ func (p *pods) Receive(ctx *actor.Context) error {
 		}
 
 	case podStatusUpdate:
-		p.receivePodStatusUpdate(ctx, msg)
+		p.receivePodStatusUpdate(ctx, msg.updatedPod)
 
 	case podEventUpdate:
 		p.receivePodEventUpdate(ctx, msg)
@@ -686,34 +686,27 @@ func (p *pods) deleteDoomedKubernetesResources(ctx *actor.Context) error {
 
 func (p *pods) startPodInformer(ctx *actor.Context) error {
 	for namespace := range p.namespaceToPoolName {
-		i := newInformer(namespace, p.podInterfaces[namespace])
-		pod, ok := i.startInformer()
+		i := newInformer(namespace, p.podInterfaces[namespace],
+			func(pod *k8sV1.Pod) { p.receivePodStatusUpdate(ctx, pod) })
+		ok := i.startInformer()
 
 		if !ok {
 			return errors.Errorf("pod informer for %s failed", namespace)
 		}
 
 		p.informers = append(p.informers, i)
-		ctx.Tell(ctx.Self(), podStatusUpdate{updatedPod: pod})
 	}
 	return nil
 }
 
 func (p *pods) startNodeInformer(ctx *actor.Context) error {
-	p.nodeInformer = newNodeInformer(p.clientSet)
-	nodeStatus, ok := p.nodeInformer.startNodeInformer()
+	p.nodeInformer = newNodeInformer(p.clientSet, p.receiveNodeStatusUpdate)
+	ok := p.nodeInformer.startNodeInformer()
 
 	if !ok {
 		return errors.Errorf("node informer failed")
 	}
 
-	if nodeStatus.updatedNode != nil {
-		p.currentNodes[nodeStatus.updatedNode.Name] = nodeStatus.updatedNode
-	}
-
-	if nodeStatus.deletedNode != nil {
-		delete(p.currentNodes, nodeStatus.deletedNode.Name)
-	}
 	return nil
 }
 
@@ -783,20 +776,30 @@ func (p *pods) receiveStartTaskPod(ctx *actor.Context, msg StartTaskPod) error {
 	return nil
 }
 
-func (p *pods) receivePodStatusUpdate(ctx *actor.Context, msg podStatusUpdate) {
-	ref, ok := p.podNameToPodHandler[msg.updatedPod.Name]
+func (p *pods) receiveNodeStatusUpdate(node *k8sV1.Node, toUpdate bool) {
+	if node != nil {
+		if toUpdate {
+			p.currentNodes[node.Name] = node
+		} else {
+			delete(p.currentNodes, node.Name)
+		}
+	}
+}
+
+func (p *pods) receivePodStatusUpdate(ctx *actor.Context, pod *k8sV1.Pod) {
+	ref, ok := p.podNameToPodHandler[pod.Name]
 	if !ok {
-		ctx.Log().WithField("pod-name", msg.updatedPod.Name).Warn(
+		ctx.Log().WithField("pod-name", pod.Name).Warn(
 			"received pod status update for un-registered pod")
 		return
 	}
 
-	ctx.Tell(ref, msg)
+	ctx.Tell(ref, podStatusUpdate{pod})
 
-	if containerID, ok := p.podNameToContainerID[msg.updatedPod.Name]; ok {
+	if containerID, ok := p.podNameToContainerID[pod.Name]; ok {
 		if state, ok := p.containerIDToSchedulingState[containerID]; ok {
 			currState := sproto.SchedulingStateQueued
-			if msg.updatedPod.Status.Phase == "Running" {
+			if pod.Status.Phase == "Running" {
 				currState = sproto.SchedulingStateScheduled
 			}
 			if currState != state {