wip fsm

Author: MiguelAAe

apis/go/mlops/scheduler/storage.pb.go

@@ -20,3 +20,8 @@ message ExperimentSnapshot {
   // on restart, which would guard against lost events in communication.
   bool deleted = 2;
 }
+
+message ModelSnapshot{
+  repeated ModelVersionStatus versions = 1;
+  bool deleted = 2;
+}

apis/mlops/scheduler/storage.proto

@@ -1,10 +1,20 @@
+/*
+Copyright (c) 2024 Seldon Technologies Ltd.
+
+Use of this software is governed BY
+(1) the license included in the LICENSE file or
+(2) if the license included in the LICENSE file is the Business Source License 1.1,
+the Change License after the Change Date as each is defined in accordance with the LICENSE file.
+*/
+
 package fsm
 
 import (
 	"context"
 	"fmt"
 
 	pb "github.com/seldonio/seldon-core/apis/go/v2/mlops/scheduler"
+	"github.com/seldonio/seldon-core/scheduler/v2/pkg/fsm/statemachine"
 )
 
 // Concrete event types
@@ -23,36 +33,29 @@ func (e *LoadModelEvent) Marshal() ([]byte, error) {
 
 type LoadModelEventHandler struct {
 	store KVStore
+	fsm   statemachine.FSM
 }
 
 func NewLoadModelEventHandler(store KVStore) *LoadModelEventHandler {
-	return &LoadModelEventHandler{store: store}
+	return &LoadModelEventHandler{store: store, fsm: statemachine.NewStateMachine()}
 }
 
 // Handle implementations (business logic goes here)
 func (e *LoadModelEventHandler) Handle(ctx context.Context, event Event) ([]OutputEvent, error) {
 	loadEvent, ok := event.(*LoadModelEvent)
 	if !ok {
-		return nil, fmt.Errorf("invalid event type")
+		return nil, fmt.Errorf("invalid event type, expected LoadModelEvent")
 	}
 
-	// save new model to storage if new
-
-	// if not new update model storage
-
-	// generate model event status for other services to ingest
+	// 1. Load current cluster state from KVStore
 
-	// schedule model into server
+	// 2. Call pure business logic from state machine statemachine.ApplyLoadModel
 
-	// retrieve servers and their models
+	// 3. Get resulting state in cluster
 
-	// filter out the server
+	// 4. convert domain events into infrastructure events
 
-	// find a suitable server to schedule
-
-	// generate agent schedule event
-
-	// generate event to update pipeline status if pipeline is affected by model
+	// 5. save cluster state and fan out events
 
 	/*
 		Errors:
@@ -69,12 +72,5 @@ func (e *LoadModelEventHandler) Handle(ctx context.Context, event Event) ([]Outp
 
 	*/
 
-	// TODO: Implement business logic
-	// - Validate request
-	// - Update model state in KVStore
-	// - Generate output events (e.g., "ModelLoadStarted", "NotifyServer", etc.)
-
-	_ = loadEvent // use the event
-
 	return []OutputEvent{}, nil
 }

scheduler/pkg/fsm/event_load_model.go

@@ -1,3 +1,12 @@
+/*
+Copyright (c) 2024 Seldon Technologies Ltd.
+
+Use of this software is governed BY
+(1) the license included in the LICENSE file or
+(2) if the license included in the LICENSE file is the Business Source License 1.1,
+the Change License after the Change Date as each is defined in accordance with the LICENSE file.
+*/
+
 package fsm
 
 import (
@@ -11,26 +20,11 @@ import (
 type KVStore interface {
 	// Get retrieves value and current version
 	Get(ctx context.Context, key string) (value []byte, version int64, err error)
-	// Set with version check (returns ErrVersionConflict if version doesn't match)
-	SetWithVersion(ctx context.Context, key string, value []byte, expectedVersion int64) (newVersion int64, err error)
 	// Set without version check (for initial writes)
 	Set(ctx context.Context, key string, value []byte) (int64, error)
 	Delete(ctx context.Context, key string) error
 }
 
-// ErrVersionConflict indicates an optimistic locking conflict
-var ErrVersionConflict = fmt.Errorf("version conflict")
-
-// EventLog interface for append-only event storage
-type EventLog interface {
-	// Append writes an event to the log and returns its sequence number
-	Append(ctx context.Context, event Event) (int64, error)
-	// MarkCommitted marks an event as successfully processed
-	MarkCommitted(ctx context.Context, seqNum int64) error
-	// GetUncommitted returns all uncommitted events (for crash recovery)
-	GetUncommitted(ctx context.Context) ([]Event, error)
-}
-
 // Event represents an incoming event to the FSM
 type Event interface {
 	Type() EventType
@@ -53,19 +47,17 @@ type OutputEvent interface {
 type Fsm struct {
 	name     string
 	store    KVStore
-	log      EventLog
 	handlers map[EventType]Handler
 }
 
 type Handler interface {
 	Handle(ctx context.Context, ev Event) ([]OutputEvent, error)
 }
 
-func NewFSM(name string, store KVStore, log EventLog) *Fsm {
+func NewFSM(name string, store KVStore) *Fsm {
 	fsm := &Fsm{
 		name:     name,
 		store:    store,
-		log:      log,
 		handlers: make(map[EventType]Handler),
 	}
 

scheduler/pkg/fsm/fsm.go

@@ -0,0 +1,10 @@
+package statemachine
+
+import pb "github.com/seldonio/seldon-core/apis/go/v2/mlops/scheduler"
+
+// ClusterState represents the state of things in the cluster needed for an event to be applied
+type ClusterState struct {
+	Model       map[string]*pb.ModelSnapshot
+	Pipelines   map[string]*pb.PipelineSnapshot
+	Experiments map[string]*pb.ExperimentSnapshot
+}