[WIP] Announcements and Monitoring of DST

cmd/dst/run.go

@@ -13,6 +13,7 @@ import (
 	"github.com/resonatehq/resonate/cmd/util"
 	"github.com/resonatehq/resonate/internal/aio"
 	"github.com/resonatehq/resonate/internal/announcements"
+	"github.com/resonatehq/resonate/internal/announcements/monitors"
 	"github.com/resonatehq/resonate/internal/api"
 	"github.com/resonatehq/resonate/internal/app/coroutines"
 	"github.com/resonatehq/resonate/internal/app/subsystems/aio/network"
@@ -73,11 +74,11 @@ func RunDSTCmd() *cobra.Command {
 			}))
 			slog.SetDefault(logger)
 
-			monitor := announcements.NewMonitor()
-			monitor.RegisterEventHandler("TaskClaimed", announcements.TaskClaimedHandler)
-			monitor.RegisterEventHandler("TaskCompleted", announcements.TaskCompletedHandler)
+			announcements.Initialize(announcements.Dst, []announcements.Monitors{})
 
-			announcements.Initialize(announcements.Dst, monitor)
+			// register monitors
+			taskMonitor := monitors.NewTaskMonitor()
+			announcements.GetInstance().Register(taskMonitor)
 
 			// instantiate metrics
 			reg := prometheus.NewRegistry()

internal/announcements/announce.go

@@ -28,34 +28,36 @@
 	e.Data[key] = value
 }
 
 func (e *Event) Get(key string) (interface{}, error) {
 	value, exists := e.Data[key]
 	if !exists {
 		return nil, fmt.Errorf("key '%s' does not exist in event of type '%s'", key, e.Type)
 	}
 	return value, nil
 }
 
 type Announcement interface {
 	Announce(event *Event)
-	Register(monitor *Monitor)
+	Register(monitor Monitors)
 }
 
 type NoopAnnouncement struct{}
 
 type DstAnnouncement struct {
 	announcements []Event
-	monitor       *Monitor
+	monitors      []Monitors
 	mutex         sync.Mutex // Mutex for thread safety
 }
 
 // Register implements Announcement.
-func (d *NoopAnnouncement) Register(monitor *Monitor) {
+func (d *NoopAnnouncement) Register(monitor Monitors) {
 	// Do nothing
 }
 
-func (d *DstAnnouncement) Register(monitor *Monitor) {
-	panic("unimplemented")
+func (d *DstAnnouncement) Register(monitor Monitors) {
+	d.mutex.Lock()
+	defer d.mutex.Unlock()
+	d.monitors = append(d.monitors, monitor)
 }
 
 var (
@@ -70,18 +72,18 @@
 	Dst
 )
 
-func Initialize(envType EnvironmentType, monitor *Monitor) {
+func Initialize(envType EnvironmentType, monitors []Monitors) {
 	once.Do(func() {
 		switch envType {
 		case Noop:
 			instance = &NoopAnnouncement{}
 		case Dst:
 			instance = &DstAnnouncement{
 				announcements: make([]Event, 0, 100), // Preallocate capacity to prevent frequent reallocations
-				monitor:       monitor,
+				monitors:      monitors,
 			}
 		default:
 			panic("Invalid environment type.")
 		}
 	})
 }
@@ -91,7 +93,7 @@
 	return instance
 }
 
 func (n *NoopAnnouncement) Announce(event *Event) {
 	// Do nothing
 }
 
@@ -101,15 +103,8 @@
 	d.announcements = append(d.announcements, *event)
 	// Print the announcement
 	fmt.Println("Announcement:", event.Type, event.Data)
-	// Apply the event to the monitor
-	d.monitor.Apply(*event)
+	// Apply the all the registered monitors
+	for _, monitor := range d.monitors {
+		monitor.Apply(*event)
+	}
 }
-
-// func (d *DstAnnouncement) GetAnnouncements() []Event {
-// 	d.mutex.Lock()
-// 	defer d.mutex.Unlock()
-// 	// Return a copy of the announcements slice to ensure thread safety
-// 	announcementsCopy := make([]Event, len(d.announcements))
-// 	copy(announcementsCopy, d.announcements)
-// 	return announcementsCopy
-// }

internal/announcements/monitor.go

@@ -1,90 +1,8 @@
-package announcements
-
-import (
-	"fmt"
-	"sync"
-)
-
-type EventHandler func(event Event, monitor *Monitor)
-
-type Monitor struct {
-	errors        []error
-	eventHandlers map[string]EventHandler // Map to store event handlers by event type
-	mutex         sync.Mutex              // Mutex for thread safety
-}
-
-func NewMonitor() *Monitor {
-	return &Monitor{
-		eventHandlers: make(map[string]EventHandler),
-	}
-}
-
-func (m *Monitor) Apply(event Event) {
-	m.mutex.Lock()
-	defer m.mutex.Unlock()
-
-	handler, ok := m.eventHandlers[event.Type]
-	if !ok {
-		m.errors = append(m.errors, fmt.Errorf("no handler found for event type: %s", event.Type))
-		return
-	}
-
-	handler(event, m)
-}
-
-func (m *Monitor) Status() []error {
-	return m.errors
-}
-
-func (m *Monitor) RegisterEventHandler(eventType string, handler EventHandler) {
-	m.mutex.Lock()
-	defer m.mutex.Unlock()
-
-	m.eventHandlers[eventType] = handler
-}
-
-// ----------------------------------------------------------------------------------
-
-// Task Cache
-type TaskCache struct {
-	tasks map[string]Event
-	mutex sync.Mutex
-}
-
-var taskCache = TaskCache{
-	tasks: make(map[string]Event),
-}
+// monitor.go
 
-// TaskClaimedHandler is an event handler for the "TaskClaimed" event type.
-// It checks if the corresponding "TaskCompleted" event is received within a certain timeframe.
-// If not, it raises an alert.
-func TaskClaimedHandler(event Event, monitor *Monitor) {
-	taskID, err := event.Get("taskId")
-	// fmt.Println("-------------------------Monitor: Task claimed:", taskID)
-	if err != nil {
-		monitor.errors = append(monitor.errors, fmt.Errorf("error getting task ID: %v", err))
-		return
-	}
-
-	// Check if the task is already in the cache
-	// If not, add it to the cache
-	if _, ok := taskCache.tasks[taskID.(string)]; !ok {
-		taskCache.mutex.Lock()
-		taskCache.tasks[taskID.(string)] = event
-		taskCache.mutex.Unlock()
-	}
-}
+package announcements
 
-func TaskCompletedHandler(event Event, monitor *Monitor) {
-	// Task completion event received, can perform any necessary actions
-	// For example, log the event or update some state
-	// fmt.Println("-------------------------Monitor: Task completed:", event)
-	if taskID, err := event.Get("taskId"); err == nil {
-		taskCache.mutex.Lock()
-		delete(taskCache.tasks, taskID.(string))
-		taskCache.mutex.Unlock()
-	} else {
-		fmt.Println("Monitor: Error getting task ID:", err)
-		monitor.errors = append(monitor.errors, fmt.Errorf("error getting task ID: %v", err))
-	}
+type Monitors interface {
+	Apply(event Event)
+	Status() []error
 }

internal/announcements/monitors/taskmonitor.go

@@ -0,0 +1,37 @@
+package monitors
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/resonatehq/resonate/internal/announcements"
+)
+
+type taskMonitor struct {
+	eventTypeCount map[string]int // Map to count occurrences of each event type
+	mutex          sync.Mutex     // Mutex for thread safety
+}
+
+func NewTaskMonitor() *taskMonitor {
+	return &taskMonitor{
+		eventTypeCount: make(map[string]int),
+	}
+}
+
+func (tm *taskMonitor) Apply(event announcements.Event) {
+	tm.mutex.Lock()
+	defer tm.mutex.Unlock()
+
+	// Increment event type count
+	tm.eventTypeCount[event.Type]++
+
+	// If the event type is TaskCreated or TaskCompleted, print the event
+	if event.Type == "TaskCreated" || event.Type == "TaskCompleted" {
+		fmt.Println("Received event:", event.Type, event.Data)
+	}
+}
+
+// Status returns any errors or status information associated with the taskMonitor.
+func (tm *taskMonitor) Status() []error {
+	return nil
+}

test/dst/dst.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/resonatehq/resonate/internal/aio"
 	"github.com/resonatehq/resonate/internal/announcements"
+	"github.com/resonatehq/resonate/internal/announcements/monitors"
 	"github.com/resonatehq/resonate/internal/api"
 	"github.com/resonatehq/resonate/internal/kernel/bus"
 	"github.com/resonatehq/resonate/internal/kernel/metadata"
@@ -65,12 +66,12 @@ func (d *DST) Run(r *rand.Rand, api api.API, aio aio.AIO, system *system.System,
 	// model
 	model := NewModel(d.config.Scenario)
 
-	monitor := announcements.NewMonitor()
-	monitor.RegisterEventHandler("TaskClaimed", announcements.TaskClaimedHandler)
-	monitor.RegisterEventHandler("TaskCompleted", announcements.TaskCompletedHandler)
-
 	// setup announcements
-	announcements.Initialize(announcements.Dst, monitor)
+	announcements.Initialize(announcements.Dst, []announcements.Monitors{})
+
+	// register monitors
+	taskMonitor := monitors.NewTaskMonitor()
+	announcements.GetInstance().Register(taskMonitor)
 
 	// add req/res
 	for _, req := range reqs {

test/dst/dst_test.go

@@ -99,26 +99,26 @@ func test(t *testing.T, scenario *Scenario) {
 	// specify reqs to enable
 	reqs := []t_api.Kind{
 		// PROMISE
-		// t_api.ReadPromise,
-		// t_api.SearchPromises,
-		// t_api.CreatePromise,
-		// t_api.CompletePromise,
-
-		// // SCHEDULE
-		// t_api.ReadSchedule,
-		// t_api.SearchSchedules,
-		// t_api.CreateSchedule,
-		// t_api.DeleteSchedule,
-
-		// // SUBSCRIPTION
-		// t_api.ReadSubscriptions,
-		// t_api.CreateSubscription,
-		// t_api.DeleteSubscription,
-
-		// // LOCK
-		// t_api.AcquireLock,
-		// t_api.HeartbeatLocks,
-		// t_api.ReleaseLock,
+		t_api.ReadPromise,
+		t_api.SearchPromises,
+		t_api.CreatePromise,
+		t_api.CompletePromise,
+
+		// SCHEDULE
+		t_api.ReadSchedule,
+		t_api.SearchSchedules,
+		t_api.CreateSchedule,
+		t_api.DeleteSchedule,
+
+		// SUBSCRIPTION
+		t_api.ReadSubscriptions,
+		t_api.CreateSubscription,
+		t_api.DeleteSubscription,
+
+		// LOCK
+		t_api.AcquireLock,
+		t_api.HeartbeatLocks,
+		t_api.ReleaseLock,
 
 		// TASK
 		t_api.ClaimTask,