Check for insufficient memory, cpu conditions and error out if necessary

events/events.go

@@ -21,23 +21,23 @@ type EventStatus struct {
 	Event     corev1.Event // copy of the raw event
 }
 
-func (p *EventStatus) String() string {
+func (e *EventStatus) String() string {
 	var buf strings.Builder
 	add := func(k, v string) {
 		if v != "" {
 			fmt.Fprintf(&buf, ", %s: %q", k, v)
 		}
 	}
-	fmt.Fprintf(&buf, "{Name: %q", p.Name)
-	add("UID", string(p.UID))
-	add("Namespace", p.Namespace)
+	fmt.Fprintf(&buf, "{Name: %q", e.Name)
+	add("UID", string(e.UID))
+	add("Namespace", e.Namespace)
 	return buf.String()
 }
 
-func (p *EventStatus) Equal(q *EventStatus) bool {
-	if p.UID != q.UID ||
-		p.Name != q.Name ||
-		p.Namespace != q.Namespace {
+func (e *EventStatus) Equal(q *EventStatus) bool {
+	if e.UID != q.UID ||
+		e.Name != q.Name ||
+		e.Namespace != q.Namespace {
 		return false
 	}
 	return true

events/status.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"os"
+	"strings"
 	"sync"
 	"time"
 
@@ -30,6 +31,8 @@ type Watcher struct {
 	currentEvent     types.UID
 }
 
+var errorMsgs = [2]string{"Insufficient memory", "Insufficient cpu"}
+
 // NewWatcher returns a Watcher on the provided client or an error.  The cancel
 // function is called when the Watcher determines an event has permanently
 // failed.  The Watcher will exit if the context provided is canceled, an error
@@ -101,7 +104,7 @@ func (w *Watcher) watch() {
 	for {
 		select {
 		case s, ok := <-w.ch:
-			if !ok || !w.updateEvent(s) {
+			if !ok || !w.displayEvent(s) {
 				return
 			}
 		case <-w.ctx.Done():
@@ -117,3 +120,21 @@ func (w *Watcher) display(format string, v ...any) {
 		fmt.Fprintf(w.stdout, timeNow()+format+"\n", v...)
 	}
 }
+
+func (w *Watcher) displayEvent(s *EventStatus) bool {
+	if w.progress {
+		log.Info(timeNow() + s.Event.String() + "\n")
+	}
+
+	message := s.Event.Message
+	for _, m := range errorMsgs {
+		// Error out if namespace is currentnamesapce and message contains predefinedcheck namespace
+		if w.currentNamespace == s.Namespace && strings.Contains(message, m) {
+			w.errCh <- fmt.Errorf("Event failed due to %s . Message: %s", s.Event.Reason, message)
+			w.cancel()
+			return false
+		}
+	}
+
+	return true
+}

topo/topo.go

@@ -26,6 +26,7 @@ import (
 	"github.com/kr/pretty"
 	topologyclientv1 "github.com/networkop/meshnet-cni/api/clientset/v1beta1"
 	topologyv1 "github.com/networkop/meshnet-cni/api/types/v1beta1"
+	"github.com/openconfig/kne/events"
 	"github.com/openconfig/kne/pods"
 	cpb "github.com/openconfig/kne/proto/controller"
 	tpb "github.com/openconfig/kne/proto/topo"
@@ -172,6 +173,15 @@ func (m *Manager) Create(ctx context.Context, timeout time.Duration) (rerr error
 			rerr = w.Cleanup(rerr)
 		}()
 	}
+	if w, err := events.NewWatcher(ctx, m.kClient, cancel); err != nil {
+		log.Warningf("Failed to start event watcher: %v", err)
+	} else {
+		w.SetProgress(m.progress)
+		defer func() {
+			cancel()
+			rerr = w.Cleanup(rerr)
+		}()
+	}
 	if err := m.checkNodeStatus(ctx, timeout); err != nil {
 		return err
 	}