feat(event): remove node on pipeline failure (#3736)

internal/daemon/common/handler_test.go

@@ -487,10 +487,15 @@ func (b *testMockBroker) RegisterPipeline(def eventlogger.Pipeline, opt ...event
 	return nil
 }
 func (b *testMockBroker) StopTimeAt(t time.Time) {}
+
 func (b *testMockBroker) RegisterNode(id eventlogger.NodeID, node eventlogger.Node, opt ...eventlogger.Option) error {
 	return nil
 }
 
+func (b *testMockBroker) RemoveNode(ctx context.Context, id eventlogger.NodeID) error {
+	return nil
+}
+
 func (b *testMockBroker) RemovePipelineAndNodes(ctx context.Context, t eventlogger.EventType, id eventlogger.PipelineID) (bool, error) {
 	return true, nil
 }

internal/event/eventer.go

@@ -59,6 +59,7 @@ type broker interface {
 	Reopen(ctx context.Context) error
 	StopTimeAt(now time.Time)
 	RegisterNode(id eventlogger.NodeID, node eventlogger.Node, opt ...eventlogger.Option) error
+	RemoveNode(ctx context.Context, id eventlogger.NodeID) error
 	SetSuccessThreshold(t eventlogger.EventType, successThreshold int) error
 	RegisterPipeline(def eventlogger.Pipeline, opt ...eventlogger.Option) error
 	RemovePipelineAndNodes(ctx context.Context, t eventlogger.EventType, id eventlogger.PipelineID) (bool, error)

internal/event/testing.go

@@ -361,6 +361,10 @@ func (b *testMockBroker) RegisterNode(id eventlogger.NodeID, node eventlogger.No
 	return nil
 }
 
+func (b *testMockBroker) RemoveNode(ctx context.Context, id eventlogger.NodeID) error {
+	return nil
+}
+
 func (b *testMockBroker) SetSuccessThreshold(t eventlogger.EventType, successThreshold int) error {
 	if b.successThresholds == nil {
 		b.successThresholds = map[eventlogger.EventType]int{}