tetragon: Allow retry logic for clone events

pkg/api/readyapi/readyapi.go

@@ -12,6 +12,10 @@ import (
 
 type MsgTetragonReady struct{}
 
+func (msg *MsgTetragonReady) Notify() bool {
+	return false
+}
+
 func (msg *MsgTetragonReady) RetryInternal(ev notify.Event, timestamp uint64) (*process.ProcessInternal, error) {
 	return nil, fmt.Errorf("Unsupported cache event MsgTetragonReady")
 }

pkg/eventcache/eventcache.go

@@ -109,13 +109,15 @@ func (ec *Cache) handleEvents() {
 			}
 		}
 
-		processedEvent := &tetragon.GetEventsResponse{
-			Event:    event.event.Encapsulate(),
-			NodeName: nodeName,
-			Time:     ktime.ToProto(event.timestamp),
-		}
+		if event.msg.Notify() {
+			processedEvent := &tetragon.GetEventsResponse{
+				Event:    event.event.Encapsulate(),
+				NodeName: nodeName,
+				Time:     ktime.ToProto(event.timestamp),
+			}
 
-		ec.server.NotifyListeners(event.msg, processedEvent)
+			ec.server.NotifyListeners(event.msg, processedEvent)
+		}
 	}
 	ec.cache = tmp
 }

pkg/grpc/exec/exec.go

@@ -67,6 +67,10 @@ type MsgExecveEventUnix struct {
 	processapi.MsgExecveEventUnix
 }
 
+func (msg *MsgExecveEventUnix) Notify() bool {
+	return true
+}
+
 func (msg *MsgExecveEventUnix) RetryInternal(ev notify.Event, timestamp uint64) (*process.ProcessInternal, error) {
 	return nil, fmt.Errorf("Unreachable state: MsgExecveEventUnix with missing internal")
 }
@@ -142,18 +146,27 @@ type MsgCloneEventUnix struct {
 	processapi.MsgCloneEvent
 }
 
+func (msg *MsgCloneEventUnix) Notify() bool {
+	return false
+}
+
 func (msg *MsgCloneEventUnix) RetryInternal(ev notify.Event, timestamp uint64) (*process.ProcessInternal, error) {
-	return nil, fmt.Errorf("Unreachable state: MsgCloneEventUnix with missing internal")
+	return nil, process.AddCloneEvent(&msg.MsgCloneEvent)
 }
 
 func (msg *MsgCloneEventUnix) Retry(internal *process.ProcessInternal, ev notify.Event) error {
-	return eventcache.HandleGenericEvent(internal, ev)
+	return nil
 }
 
 func (msg *MsgCloneEventUnix) HandleMessage() *tetragon.GetEventsResponse {
 	switch msg.Common.Op {
 	case ops.MSG_OP_CLONE:
-		process.AddCloneEvent(&msg.MsgCloneEvent)
+		if err := process.AddCloneEvent(&msg.MsgCloneEvent); err != nil {
+			ec := eventcache.Get()
+			if ec != nil {
+				ec.Add(nil, nil, msg.MsgCloneEvent.Ktime, msg)
+			}
+		}
 	default:
 		logger.GetLogger().WithField("message", msg).Warn("HandleCloneMessage: Unhandled event")
 	}
@@ -207,6 +220,10 @@ type MsgExitEventUnix struct {
 	tetragonAPI.MsgExitEvent
 }
 
+func (msg *MsgExitEventUnix) Notify() bool {
+	return true
+}
+
 func (msg *MsgExitEventUnix) RetryInternal(ev notify.Event, timestamp uint64) (*process.ProcessInternal, error) {
 	p := ev.GetProcess()
 	internal, parent := process.GetParentProcessInternal(p.Pid.Value, timestamp)

pkg/grpc/test/test.go

@@ -22,6 +22,10 @@ type MsgTestEventUnix struct {
 	testapi.MsgTestEvent
 }
 
+func (msg *MsgTestEventUnix) Notify() bool {
+	return true
+}
+
 func (msg *MsgTestEventUnix) RetryInternal(ev notify.Event, timestamp uint64) (*process.ProcessInternal, error) {
 	return eventcache.HandleGenericInternal(ev, timestamp)
 }

pkg/grpc/tracing/tracing.go

@@ -193,6 +193,10 @@ type MsgGenericTracepointUnix struct {
 	Args       []tracingapi.MsgGenericTracepointArg
 }
 
+func (msg *MsgGenericTracepointUnix) Notify() bool {
+	return true
+}
+
 func (msg *MsgGenericTracepointUnix) RetryInternal(ev notify.Event, timestamp uint64) (*process.ProcessInternal, error) {
 	return eventcache.HandleGenericInternal(ev, timestamp)
 }
@@ -282,6 +286,10 @@ type MsgGenericKprobeUnix struct {
 	Args         []tracingapi.MsgGenericKprobeArg
 }
 
+func (msg *MsgGenericKprobeUnix) Notify() bool {
+	return true
+}
+
 func (msg *MsgGenericKprobeUnix) RetryInternal(ev notify.Event, timestamp uint64) (*process.ProcessInternal, error) {
 	return eventcache.HandleGenericInternal(ev, timestamp)
 }

pkg/process/process.go

@@ -268,12 +268,12 @@ func AddExecEvent(event *tetragonAPI.MsgExecveEventUnix) *ProcessInternal {
 }
 
 // AddCloneEvent adds a new process into the cache from a CloneEvent
-func AddCloneEvent(event *tetragonAPI.MsgCloneEvent) {
+func AddCloneEvent(event *tetragonAPI.MsgCloneEvent) error {
 	parentExecId := GetProcessID(event.Parent.Pid, event.Parent.Ktime)
 	parent, err := Get(parentExecId)
 	if err != nil {
 		logger.GetLogger().WithField("parent-exec-id", parentExecId).Debug("AddCloneEvent: process not found in cache")
-		return
+		return err
 	}
 	pi := parent.GetProcessInternalCopy()
 	if pi.process != nil {
@@ -288,6 +288,7 @@ func AddCloneEvent(event *tetragonAPI.MsgCloneEvent) {
 		}
 	}
 	procCache.Add(pi)
+	return nil
 }
 
 func Get(execId string) (*ProcessInternal, error) {

pkg/reader/notify/notify.go

@@ -12,6 +12,7 @@ type Message interface {
 	HandleMessage() *tetragon.GetEventsResponse
 	RetryInternal(Event, uint64) (*process.ProcessInternal, error)
 	Retry(*process.ProcessInternal, Event) error
+	Notify() bool
 }
 
 type Event interface {