[WIP] Announcements and Monitoring of DST

cmd/dst/run.go

@@ -12,6 +12,8 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"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"
@@ -72,6 +74,12 @@ func RunDSTCmd() *cobra.Command {
 			}))
 			slog.SetDefault(logger)
 
+			announcements.Initialize(announcements.Dst, []announcements.Monitors{})
+
+			// register monitors
+			taskMonitor := monitors.NewTaskMonitor()
+			announcements.GetInstance().Register(taskMonitor)
+
 			// instantiate metrics
 			reg := prometheus.NewRegistry()
 			metrics := metrics.New(reg)

internal/announcements/announce.go

@@ -0,0 +1,110 @@
+package announcements
+
+import (
+	"fmt"
+	"sync"
+)
+
+type Event struct {
+	Type string
+	Data map[string]interface{}
+}
+
+func NewEvent(eventType string, initialData ...map[string]interface{}) *Event {
+	var data map[string]interface{}
+	if len(initialData) > 0 {
+		data = initialData[0] // Use the first map provided if any.
+	} else {
+		data = make(map[string]interface{})
+	}
+
+	return &Event{
+		Type: eventType,
+		Data: data,
+	}
+}
+
+func (e *Event) Set(key string, value interface{}) {
+	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 Monitors)
+}
+
+type NoopAnnouncement struct{}
+
+type DstAnnouncement struct {
+	announcements []Event
+	monitors      []Monitors
+	mutex         sync.Mutex // Mutex for thread safety
+}
+
+// Register implements Announcement.
+func (d *NoopAnnouncement) Register(monitor Monitors) {
+	// Do nothing
+}
+
+func (d *DstAnnouncement) Register(monitor Monitors) {
+	d.mutex.Lock()
+	defer d.mutex.Unlock()
+	d.monitors = append(d.monitors, monitor)
+}
+
+var (
+	instance Announcement
+	once     sync.Once
+)
+
+type EnvironmentType int
+
+const (
+	Noop EnvironmentType = iota
+	Dst
+)
+
+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
+				monitors:      monitors,
+			}
+		default:
+			panic("Invalid environment type.")
+		}
+	})
+}
+
+func GetInstance() Announcement {
+	// check if the instance has been initialized
+	return instance
+}
+
+func (n *NoopAnnouncement) Announce(event *Event) {
+	// Do nothing
+}
+
+func (d *DstAnnouncement) Announce(event *Event) {
+	d.mutex.Lock()
+	defer d.mutex.Unlock()
+	d.announcements = append(d.announcements, *event)
+	// Print the announcement
+	fmt.Println("Announcement:", event.Type, event.Data)
+	// Apply the all the registered monitors
+	for _, monitor := range d.monitors {
+		monitor.Apply(*event)
+	}
+}

internal/announcements/monitor.go

@@ -0,0 +1,8 @@
+// monitor.go
+
+package announcements
+
+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
+}

internal/app/subsystems/aio/network/network_dst.go

@@ -5,6 +5,7 @@ import (
 	"net/http"
 
 	"github.com/resonatehq/resonate/internal/aio"
+	"github.com/resonatehq/resonate/internal/announcements"
 	"github.com/resonatehq/resonate/internal/kernel/bus"
 	"github.com/resonatehq/resonate/internal/kernel/t_aio"
 	"github.com/resonatehq/resonate/internal/util"
@@ -69,10 +70,22 @@ func (d *NetworkDSTDevice) Process(sqes []*bus.SQE[t_aio.Submission, t_aio.Compl
 				res = &http.Response{
 					StatusCode: http.StatusOK,
 				}
+
+				event := announcements.NewEvent("HTTPResponse")
+				event.Set("StatusCode", res.StatusCode)
+				event.Set("URL", sqe.Submission.Network.Http.Url)
+				event.Set("Method", sqe.Submission.Network.Http.Method)
+				announcements.GetInstance().Announce(event)
 			} else {
 				res = &http.Response{
 					StatusCode: http.StatusInternalServerError,
 				}
+
+				event := announcements.NewEvent("HTTPResponse")
+				event.Set("StatusCode", res.StatusCode)
+				event.Set("URL", sqe.Submission.Network.Http.Url)
+				event.Set("Method", sqe.Submission.Network.Http.Method)
+				announcements.GetInstance().Announce(event)
 			}
 
 			cqe.Completion = &t_aio.Completion{

internal/kernel/t_aio/network.go

@@ -17,6 +17,7 @@ type NetworkSubmission struct {
 }
 
 func (s *NetworkSubmission) String() string {
+
 	switch s.Kind {
 	case Http:
 		return fmt.Sprintf("Network(http=Http(method=%s, url=%s))", s.Http.Method, s.Http.Url)

test/dst/dst.go

@@ -7,6 +7,8 @@ import (
 	"strconv"
 
 	"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"
@@ -64,6 +66,13 @@ func (d *DST) Run(r *rand.Rand, api api.API, aio aio.AIO, system *system.System,
 	// model
 	model := NewModel(d.config.Scenario)
 
+	// setup announcements
+	announcements.Initialize(announcements.Dst, []announcements.Monitors{})
+
+	// register monitors
+	taskMonitor := monitors.NewTaskMonitor()
+	announcements.GetInstance().Register(taskMonitor)
+
 	// add req/res
 	for _, req := range reqs {
 		switch req {

test/dst/generator.go

@@ -6,6 +6,7 @@ import (
 	"math/rand" // nosemgrep
 	"strconv"
 
+	"github.com/resonatehq/resonate/internal/announcements"
 	"github.com/resonatehq/resonate/internal/kernel/t_api"
 	"github.com/resonatehq/resonate/pkg/idempotency"
 	"github.com/resonatehq/resonate/pkg/promise"
@@ -396,6 +397,11 @@ func (g *Generator) GenerateClaimTask(r *rand.Rand, t int64) *t_api.Request {
 	executionId := g.idSet[r.Intn(len(g.idSet))]
 	expiryInMilliseconds := RangeInt63n(r, t, g.ticks*g.timeElapsedPerTick)
 
+	// claim task event announcement
+	event := announcements.NewEvent("TaskClaimed", map[string]interface{}{"taskId": taskId, "counter": counter, "processId": processId, "executionId": executionId, "expiryInMilliseconds": expiryInMilliseconds})
+
+	announcements.GetInstance().Announce(event)
+
 	return &t_api.Request{
 		Kind: t_api.ClaimTask,
 		ClaimTask: &t_api.ClaimTaskRequest{
@@ -420,6 +426,10 @@ func (g *Generator) GenerateCompleteTask(r *rand.Rand, t int64) *t_api.Request {
 		state = promise.Rejected
 	}
 
+	// complete task event announcement
+	event := announcements.NewEvent("TaskCompleted", map[string]interface{}{"taskId": taskId, "counter": counter, "executionId": executionId, "state": state, "headers": headers, "data": data})
+	announcements.GetInstance().Announce(event)
+
 	return &t_api.Request{
 		Kind: t_api.CompleteTask,
 		CompleteTask: &t_api.CompleteTaskRequest{