Implement state change event

evidolob · evidolob · commit a067a0fa193e · 2023-11-01T14:16:22.000+02:00
Signed-off-by: Yevhen Vydolob &lt;yvydolob@redhat.com&gt;
diff --git a/pkg/crc/api/events/cluster_load_stream.go b/pkg/crc/api/events/cluster_load_stream.go
@@ -20,7 +20,7 @@ type TickListener struct {
 }
 
 func newClusterLoadStream(server *EventServer) EventStream {
-	return newStream(newStatusListener(server.machine), newEventPublisher(CLUSTER_LOAD, server.sseServer))
+	return newStream(newStatusListener(server.machine), newEventPublisher(ClusterLoad, server.sseServer))
 }
 
 func newStatusListener(machine crcMachine.Client) EventProducer {
diff --git a/pkg/crc/api/events/event_server.go b/pkg/crc/api/events/event_server.go
@@ -54,8 +54,9 @@ func NewEventServer(machine machine.Client) *EventServer {
 		stream.RemoveSubscriber(sub)
 	}
 
-	sseServer.CreateStream(LOGS)
-	sseServer.CreateStream(CLUSTER_LOAD)
+	sseServer.CreateStream(Logs)
+	sseServer.CreateStream(ClusterLoad)
+	sseServer.CreateStream(StatusChange)
 	return eventServer
 }
 
@@ -65,10 +66,12 @@ func (es *EventServer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 
 func createEventStream(server *EventServer, streamID string) EventStream {
 	switch streamID {
-	case LOGS:
+	case Logs:
 		return newLogsStream(server)
-	case CLUSTER_LOAD:
+	case ClusterLoad:
 		return newClusterLoadStream(server)
+	case StatusChange:
+		return newStatusChangeStream(server)
 	}
 	return nil
 }
diff --git a/pkg/crc/api/events/events.go b/pkg/crc/api/events/events.go
@@ -3,8 +3,9 @@ package events
 import "github.com/r3labs/sse/v2"
 
 const (
-	LOGS         = "logs"         // Logs event channel, contains daemon logs
-	CLUSTER_LOAD = "cluster_load" // status event channel, contains VM load info
+	Logs         = "logs"          // Logs event channel, contains daemon logs
+	ClusterLoad  = "cluster_load"  // status event channel, contains VM load info
+	StatusChange = "status_change" // status change channel, fires on 'starting', 'stopping', etc
 )
 
 type EventPublisher interface {
diff --git a/pkg/crc/api/events/log_stream.go b/pkg/crc/api/events/log_stream.go
@@ -23,7 +23,7 @@ func newSSEStreamHook(server *sse.Server) *streamHook {
 		&logrus.JSONFormatter{
 			TimestampFormat:   "",
 			DisableTimestamp:  false,
-			DisableHTMLEscape: false,
+			DisableHTMLEscape: true,
 			DataKey:           "",
 			FieldMap:          nil,
 			CallerPrettyfier:  nil,
@@ -56,7 +56,12 @@ func (s *streamHook) Fire(entry *logrus.Entry) error {
 		return err
 	}
 
-	s.server.Publish(LOGS, &sse.Event{Event: []byte(LOGS), Data: line})
+	// remove "Line Feed"("\n") character which add was added by json.Encoder
+	if line[len(line)-1] == 10 {
+		line = line[:len(line)-1]
+	}
+
+	s.server.Publish(Logs, &sse.Event{Event: []byte(Logs), Data: line})
 	return nil
 }
 
diff --git a/pkg/crc/api/events/status_change_stream.go b/pkg/crc/api/events/status_change_stream.go
@@ -0,0 +1,62 @@
+package events
+
+import (
+	"encoding/json"
+
+	"github.com/crc-org/crc/v2/pkg/crc/logging"
+	"github.com/crc-org/crc/v2/pkg/crc/machine"
+	"github.com/crc-org/crc/v2/pkg/crc/machine/state"
+	"github.com/crc-org/crc/v2/pkg/crc/machine/types"
+	"github.com/crc-org/crc/v2/pkg/events"
+	"github.com/r3labs/sse/v2"
+)
+
+type statusChangeEvent struct {
+	Status *types.ClusterStatusResult `json:"status"`
+	Error  string                     `json:"error,omitempty"`
+}
+
+type statusChange struct {
+	listenerDisposable events.Disposable
+	machineClient      machine.Client
+}
+
+func newStatusChangeStream(server *EventServer) EventStream {
+	return newStream(newStatusChangeListener(server.machine), newEventPublisher(StatusChange, server.sseServer))
+}
+
+func newStatusChangeListener(client machine.Client) EventProducer {
+	return &statusChange{
+		machineClient: client,
+	}
+}
+
+func (st *statusChange) Start(publisher EventPublisher) {
+	st.listenerDisposable = events.StatusChanged.AddListener(func(changedEvent events.StatusChangedEvent) {
+		logging.Debugf("State Changed Event %s", changedEvent)
+		var event statusChangeEvent
+		status, err := st.machineClient.Status()
+		// if we cannot receive actual state, send error state with error description
+		if err != nil {
+			event = statusChangeEvent{Status: &types.ClusterStatusResult{
+				CrcStatus: state.Error,
+			}, Error: err.Error()}
+		} else {
+			status.CrcStatus = changedEvent.State // override with actual reported state
+			event = statusChangeEvent{Status: status}
+			if changedEvent.Error != nil {
+				event.Error = changedEvent.Error.Error()
+			}
+
+		}
+		data, _ := json.Marshal(event)
+		publisher.Publish(&sse.Event{Event: []byte(StatusChange), Data: data})
+	})
+}
+
+func (st *statusChange) Stop() {
+	if st.listenerDisposable != nil {
+		st.listenerDisposable()
+		st.listenerDisposable = nil
+	}
+}
diff --git a/pkg/crc/machine/sync.go b/pkg/crc/machine/sync.go
@@ -10,6 +10,7 @@ import (
 	"github.com/crc-org/crc/v2/pkg/crc/machine/state"
 	"github.com/crc-org/crc/v2/pkg/crc/machine/types"
 	crcPreset "github.com/crc-org/crc/v2/pkg/crc/preset"
+	"github.com/crc-org/crc/v2/pkg/events"
 )
 
 const startCancelTimeout = 15 * time.Second
@@ -69,6 +70,10 @@ func (s *Synchronized) Delete() error {
 
 	err := s.underlying.Delete()
 	s.syncOperationDone <- Deleting
+
+	if err == nil {
+		events.StatusChanged.Fire(events.StatusChangedEvent{State: state.Novm})
+	}
 	return err
 }
 
@@ -80,6 +85,7 @@ func (s *Synchronized) prepareStart(startCancel context.CancelFunc) error {
 	}
 	s.startCancel = startCancel
 	s.currentState = Starting
+	events.StatusChanged.Fire(events.StatusChangedEvent{State: state.Starting})
 
 	return nil
 }
@@ -92,6 +98,13 @@ func (s *Synchronized) Start(ctx context.Context, startConfig types.StartConfig)
 
 	startResult, err := s.underlying.Start(ctx, startConfig)
 	s.syncOperationDone <- Starting
+
+	if err == nil {
+		events.StatusChanged.Fire(events.StatusChangedEvent{State: startResult.Status})
+	} else {
+		events.StatusChanged.Fire(events.StatusChangedEvent{State: state.Error, Error: err})
+	}
+
 	return startResult, err
 }
 
@@ -136,10 +149,16 @@ func (s *Synchronized) Stop() (state.State, error) {
 	if err := s.prepareStopDelete(Stopping); err != nil {
 		return state.Error, err
 	}
+	events.StatusChanged.Fire(events.StatusChangedEvent{State: state.Stopping})
 
 	st, err := s.underlying.Stop()
 	s.syncOperationDone <- Stopping
 
+	if err == nil {
+		events.StatusChanged.Fire(events.StatusChangedEvent{State: st})
+	} else {
+		events.StatusChanged.Fire(events.StatusChangedEvent{State: state.Error, Error: err})
+	}
 	return st, err
 }
 
@@ -160,7 +179,14 @@ func (s *Synchronized) ConnectionDetails() (*types.ConnectionDetails, error) {
 }
 
 func (s *Synchronized) PowerOff() error {
-	return s.underlying.PowerOff()
+	err := s.underlying.PowerOff()
+	if err != nil {
+		events.StatusChanged.Fire(events.StatusChangedEvent{State: state.Stopped})
+	} else {
+		events.StatusChanged.Fire(events.StatusChangedEvent{State: state.Error, Error: err})
+	}
+
+	return err
 }
 
 func (s *Synchronized) Status() (*types.ClusterStatusResult, error) {
diff --git a/pkg/events/emitter.go b/pkg/events/emitter.go
@@ -49,6 +49,6 @@ func (e *event[T]) Fire(event T) {
 	e.eventMutex.Lock()
 	defer e.eventMutex.Unlock()
 	for _, l := range e.listeners {
-		l.Listener(event)
+		go l.Listener(event)
 	}
 }
diff --git a/pkg/events/events.go b/pkg/events/events.go
@@ -0,0 +1,14 @@
+package events
+
+import (
+	"github.com/crc-org/crc/v2/pkg/crc/machine/state"
+)
+
+type StatusChangedEvent struct {
+	State state.State
+	Error error
+}
+
+var (
+	StatusChanged = NewEvent[StatusChangedEvent]()
+)

Original file line number	Diff line number	Diff line change
`@@ -20,7 +20,7 @@ type TickListener struct {`
`20`	`20`	`}`
`21`	`21`
`22`	`22`	`func newClusterLoadStream(server *EventServer) EventStream {`
`23`		`- return newStream(newStatusListener(server.machine), newEventPublisher(CLUSTER_LOAD, server.sseServer))`
	`23`	`+ return newStream(newStatusListener(server.machine), newEventPublisher(ClusterLoad, server.sseServer))`
`24`	`24`	`}`
`25`	`25`
`26`	`26`	`func newStatusListener(machine crcMachine.Client) EventProducer {`
Original file line number	Diff line number	Diff line change
`@@ -54,8 +54,9 @@ func NewEventServer(machine machine.Client) *EventServer {`
`54`	`54`	`stream.RemoveSubscriber(sub)`
`55`	`55`	`}`
`56`	`56`
`57`		`- sseServer.CreateStream(LOGS)`
`58`		`- sseServer.CreateStream(CLUSTER_LOAD)`
	`57`	`+ sseServer.CreateStream(Logs)`
	`58`	`+ sseServer.CreateStream(ClusterLoad)`
	`59`	`+ sseServer.CreateStream(StatusChange)`
`59`	`60`	`return eventServer`
`60`	`61`	`}`
`61`	`62`
`@@ -65,10 +66,12 @@ func (es EventServer) ServeHTTP(w http.ResponseWriter, r http.Request) {`
`65`	`66`
`66`	`67`	`func createEventStream(server *EventServer, streamID string) EventStream {`
`67`	`68`	`switch streamID {`
`68`		`- case LOGS:`
	`69`	`+ case Logs:`
`69`	`70`	`return newLogsStream(server)`
`70`		`- case CLUSTER_LOAD:`
	`71`	`+ case ClusterLoad:`
`71`	`72`	`return newClusterLoadStream(server)`
	`73`	`+ case StatusChange:`
	`74`	`+ return newStatusChangeStream(server)`
`72`	`75`	`}`
`73`	`76`	`return nil`
`74`	`77`	`}`
Original file line number	Diff line number	Diff line change
`@@ -49,6 +49,6 @@ func (e *event[T]) Fire(event T) {`
`49`	`49`	`e.eventMutex.Lock()`
`50`	`50`	`defer e.eventMutex.Unlock()`
`51`	`51`	`for _, l := range e.listeners {`
`52`		`- l.Listener(event)`
	`52`	`+ go l.Listener(event)`
`53`	`53`	`}`
`54`	`54`	`}`