Add a K8s readiness check for device pods.

topo/node/arista/arista.go

@@ -177,7 +177,12 @@ func (n *Node) Status(ctx context.Context) (node.Status, error) {
 	case corev1.PodPending:
 		return node.StatusPending, nil
 	case corev1.PodRunning:
-		return node.StatusRunning, nil
+		for _, cond := range p[0].Status.Conditions {
+			if cond.Type == corev1.PodReady && cond.Status == corev1.ConditionTrue {
+				return node.StatusRunning, nil
+			}
+		}
+		return node.StatusPending, nil
 	case corev1.PodFailed:
 		return node.StatusFailed, nil
 	default:

topo/node/arista/arista_test.go

@@ -558,11 +558,12 @@ func TestResetCfg(t *testing.T) {
 
 func TestStatus(t *testing.T) {
 	tests := []struct {
-		desc      string
-		cantWatch bool
-		noPodYet  bool
-		phase     corev1.PodPhase
-		status    node.Status
+		desc       string
+		cantWatch  bool
+		noPodYet   bool
+		phase      corev1.PodPhase
+		conditions []corev1.PodCondition
+		status     node.Status
 	}{
 		{
 			desc:      "can't watch pod status",
@@ -581,10 +582,24 @@ func TestStatus(t *testing.T) {
 			status: node.StatusPending,
 		},
 		{
-			desc:   "pod running",
+			desc:   "pod running not ready",
 			phase:  corev1.PodRunning,
+			status: node.StatusPending,
+		},
+		{
+			desc:  "pod running and ready",
+			phase: corev1.PodRunning,
+			conditions: []corev1.PodCondition{{
+				Type:   corev1.PodReady,
+				Status: corev1.ConditionTrue,
+			}},
 			status: node.StatusRunning,
 		},
+		{
+			desc:   "pod failed",
+			phase:  corev1.PodFailed,
+			status: node.StatusFailed,
+		},
 	}
 
 	ctx := context.Background()
@@ -602,7 +617,8 @@ func TestStatus(t *testing.T) {
 						Namespace: ns,
 					},
 					Status: corev1.PodStatus{
-						Phase: tt.phase,
+						Phase:      tt.phase,
+						Conditions: tt.conditions,
 					},
 				})
 			}

topo/node/ciena/ciena.go

@@ -238,7 +238,8 @@ func (n *Node) CreatePod(ctx context.Context) error {
 				SecurityContext: &corev1.SecurityContext{
 					Privileged: ptr.To(true),
 				},
-				VolumeMounts: extraMounts,
+				ReadinessProbe:  node.ServiceReadinessProbe(pb),
+				VolumeMounts:    extraMounts,
 			}},
 			Volumes:                       extraVolumes,
 			TerminationGracePeriodSeconds: ptr.To[int64](0),

topo/node/ciena/ciena_test.go

@@ -25,7 +25,9 @@ import (
 	"github.com/openconfig/kne/topo/node"
 	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/anypb"
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 	kfake "k8s.io/client-go/kubernetes/fake"
 )
 
@@ -305,6 +307,12 @@ func TestNode_CreatePod_EquipmentFile(t *testing.T) {
 			Image:      "vrnetlab/ciena_waverouter:config",
 			VendorData: vendorData,
 		},
+		Services: map[uint32]*tpb.Service{
+			22: {
+				Names:  []string{"ssh"},
+				Inside: 22,
+			},
+		},
 	}
 
 	n := &Node{
@@ -337,6 +345,19 @@ func TestNode_CreatePod_EquipmentFile(t *testing.T) {
 	if ctr.Image != "vrnetlab/ciena_waverouter:config" {
 		t.Errorf("container image: got %q, want %q", ctr.Image, "vrnetlab/ciena_waverouter:config")
 	}
+	wantProbe := &corev1.Probe{
+		ProbeHandler: corev1.ProbeHandler{
+			TCPSocket: &corev1.TCPSocketAction{
+				Port: intstr.FromInt(22),
+			},
+		},
+		InitialDelaySeconds: 10,
+		PeriodSeconds:       10,
+		FailureThreshold:    60,
+	}
+	if diff := cmp.Diff(wantProbe, ctr.ReadinessProbe); diff != "" {
+		t.Errorf("container readiness probe mismatch (-want +got):\n%s", diff)
+	}
 	foundMount := false
 	for _, m := range ctr.VolumeMounts {
 		if m.Name == "equipment-file" && m.MountPath == "/equipment/setup.json" && m.SubPath == "setup.json" && m.ReadOnly {

topo/node/cisco/cisco.go

@@ -114,7 +114,10 @@ var (
 	}
 )
 
-var podIsUpRegex = regexp.MustCompile(`Router up`)
+var (
+	podIsUpRegex     = regexp.MustCompile(`Router up|Vxr up`)
+	podIsFailedRegex = regexp.MustCompile(`Router failed to come up|FATAL sim:|LoginTimeoutError`)
+)
 
 func New(nodeImpl *node.Impl) (node.Node, error) {
 	if nodeImpl == nil {
@@ -180,6 +183,7 @@ func (n *Node) Create(ctx context.Context) error {
 		tty = true
 		stdin = true
 	}
+	readinessProbe := node.ServiceReadinessProbe(pb)
 	pod := &corev1.Pod{
 		ObjectMeta: metav1.ObjectMeta{
 			Name: n.Name(),
@@ -200,6 +204,7 @@ func (n *Node) Create(ctx context.Context) error {
 				Resources:       node.ToResourceRequirements(pb.Constraints),
 				ImagePullPolicy: "IfNotPresent",
 				SecurityContext: secContext,
+				ReadinessProbe:  readinessProbe,
 				VolumeMounts: []corev1.VolumeMount{{
 					Name:      fmt.Sprintf("%s-run-mount", pb.Name),
 					ReadOnly:  false,
@@ -610,7 +615,8 @@ func defaults(pb *tpb.Node) (*tpb.Node, error) {
 }
 
 // Status returns the current node state.
-// For 8000e nodes it checks the logs and return running if log contains "Router up"
+// For 8000e nodes it checks the logs and return running if log contains "Router up" and pod is ready.
+// For XRD nodes it returns running if pod is ready.
 func (n *Node) Status(ctx context.Context) (node.Status, error) {
 	p, err := n.Pods(ctx)
 	if err != nil {
@@ -630,36 +636,57 @@ func (n *Node) Status(ctx context.Context) (node.Status, error) {
 		pb := n.Proto
 		if pb.GetModel() != ModelXRD {
 			req := n.KubeClient.CoreV1().Pods(p[0].Namespace).GetLogs(p[0].Name, &corev1.PodLogOptions{})
-			if !isNode8000eUp(ctx, req) {
+			logBytes, err := getPodLogs(ctx, req)
+			if err != nil {
 				log.V(2).Infof("Cisco %s node %s status is %v", n.Proto.Model, n.Name(), node.StatusPending)
 				return node.StatusPending, nil
 			}
-		}
-		for _, cond := range p[0].Status.Conditions {
-			if cond.Type == corev1.PodReady && cond.Status != corev1.ConditionTrue {
+			if isNode8000eFailed(logBytes) {
+				log.Warningf("Cisco %s node %s failed to boot", n.Proto.Model, n.Name())
+				return node.StatusFailed, fmt.Errorf("cisco %s node %s failed to boot", n.Proto.Model, n.Name())
+			}
+			if !isNode8000eUp(logBytes) {
 				log.V(2).Infof("Cisco %s node %s status is %v", n.Proto.Model, n.Name(), node.StatusPending)
 				return node.StatusPending, nil
 			}
 		}
-		log.Infof("Cisco %s node %s status is %v ", n.Proto.Model, n.Name(), node.StatusRunning)
-		return node.StatusRunning, nil
+		for _, cond := range p[0].Status.Conditions {
+			if cond.Type == corev1.PodReady && cond.Status == corev1.ConditionTrue {
+				log.Infof("Cisco %s node %s status is %v ", n.Proto.Model, n.Name(), node.StatusRunning)
+				return node.StatusRunning, nil
+			}
+		}
+		log.V(2).Infof("Cisco %s node %s status is %v", n.Proto.Model, n.Name(), node.StatusPending)
+		return node.StatusPending, nil
 	default:
 		return node.StatusUnknown, nil
 	}
 }
 
-func isNode8000eUp(ctx context.Context, req *rest.Request) bool {
+func getPodLogs(ctx context.Context, req *rest.Request) ([]byte, error) {
 	podLogs, err := req.Stream(ctx)
 	if err != nil {
-		return false
+		return nil, err
 	}
-	defer podLogs.Close()
+	defer func() {
+		if err := podLogs.Close(); err != nil {
+			log.V(2).Infof("Failed to close pod logs stream: %v", err)
+		}
+	}()
 	buf := new(bytes.Buffer)
-	len, err := io.Copy(buf, podLogs)
-	if err != nil || len == 0 {
-		return false
+	n, err := io.Copy(buf, podLogs)
+	if err != nil || n == 0 {
+		return nil, fmt.Errorf("failed to read pod logs or empty log")
 	}
-	return podIsUpRegex.Match(buf.Bytes())
+	return buf.Bytes(), nil
+}
+
+func isNode8000eUp(logBytes []byte) bool {
+	return podIsUpRegex.Match(logBytes)
+}
+
+func isNode8000eFailed(logBytes []byte) bool {
+	return podIsFailedRegex.Match(logBytes)
 }
 
 // No op function to override default network on open function.