Add Platform Router and Triton implementation

example/e2e/system.yaml

@@ -8,7 +8,7 @@ pipeline:
   start:
     aerospike:
       action: docker:run
-      image: 'aerospike/aerospike-server'
+      image: 'aerospike/aerospike-server:3.16.0.6'
       name: mly_aero
       ports:
         3000: 3000

service/config/model.go

@@ -17,6 +17,10 @@ type Model struct {
 	ID    string
 	Debug bool
 
+	// Platform specifies the model platform (tensorflow, triton)
+	// Defaults to "tensorflow" for backward compatibility
+	Platform string `json:",omitempty" yaml:",omitempty"`
+
 	// Location is the path the model will be copied to.
 	Location string `json:",omitempty" yaml:",omitempty"`
 
@@ -52,6 +56,9 @@ type Model struct {
 	// All requests are eligible to be logged.
 	Stream *config.Stream `json:",omitempty" yaml:",omitempty"`
 
+	// Triton configuration for Triton Inference Server models
+	Triton *TritonConfig `json:",omitempty" yaml:",omitempty"`
+
 	// Modified shows the state of the model files.
 	Modified *Modified `json:",omitempty" yaml:",omitempty"`
 
@@ -116,15 +123,60 @@ func (m *Model) Init(globalBatchConfig *batchconfig.BatcherConfig) {
 	}
 }
 
-// Validate validates model config
 func (m *Model) Validate() error {
 	if m.ID == "" {
 		return fmt.Errorf("model.ID was empty")
 	}
 
-	if m.URL == "" {
-		return fmt.Errorf("model.URL was empty")
+	// Platform-specific validation
+	platform := m.GetPlatform()
+	switch platform {
+	case "tensorflow":
+		if m.URL == "" {
+			return fmt.Errorf("tensorflow model %s requires URL", m.ID)
+		}
+	case "triton":
+		// Triton models require Triton configuration
+		if m.Triton == nil {
+			return fmt.Errorf("triton model %s requires Triton configuration", m.ID)
+		}
+		if m.URL == "" {
+			return fmt.Errorf("triton model %s requires URL (Triton server endpoint)", m.ID)
+		}
+		if err := m.Triton.Validate(); err != nil {
+			return fmt.Errorf("triton model %s config invalid: %w", m.ID, err)
+		}
+	default:
+		return fmt.Errorf("unsupported platform '%s' for model %s (supported: tensorflow, triton)", platform, m.ID)
 	}
 
 	return nil
 }
+
+// TritonConfig represents Triton Inference Server specific configuration
+type TritonConfig struct {
+	ModelName string `json:",omitempty" yaml:",omitempty"` // Model name in Triton
+	Version   string `json:",omitempty" yaml:",omitempty"` // Model version (defaults to "1")
+	Timeout   int    `json:",omitempty" yaml:",omitempty"` // HTTP timeout in milliseconds
+}
+
+func (t *TritonConfig) Validate() error {
+	if t.ModelName == "" {
+		return fmt.Errorf("Triton ModelName is required")
+	}
+	if t.Version == "" {
+		t.Version = "1" // Default version
+	}
+	if t.Timeout <= 0 {
+		t.Timeout = 100
+	}
+	return nil
+}
+
+// GetPlatform returns the platform with default to "tensorflow" for backward compatibility
+func (m *Model) GetPlatform() string {
+	if m.Platform == "" {
+		return "tensorflow"
+	}
+	return m.Platform
+}

service/endpoint/model.go

@@ -16,7 +16,6 @@ import (
 	"github.com/viant/mly/service/config"
 	serviceConfig "github.com/viant/mly/service/config"
 	"github.com/viant/mly/service/endpoint/meta"
-	"github.com/viant/mly/service/tfmodel"
 	"github.com/viant/mly/shared/common"
 	"github.com/viant/mly/shared/datastore"
 	"golang.org/x/sync/semaphore"
@@ -106,9 +105,27 @@ func Build(mux *http.ServeMux, config *Config, datastores map[string]*datastore.
 			mstart := time.Now()
 
 			log.Printf("[%s] model loading", model.ID)
+
+			// Validate model configuration first
+			if validateErr := model.Validate(); validateErr != nil {
+				log.Printf("[%s] ERROR: Model validation failed: %v", model.ID, validateErr)
+				lock.Lock()
+				err = fmt.Errorf("model %s validation failed: %w", model.ID, validateErr)
+				lock.Unlock()
+				return
+			}
+			log.Printf("[%s] Model configuration validated successfully", model.ID)
+
 			e := func() error {
-				tfService := tfmodel.NewService(model, fs, metrics, sema, cfge.MaxEvaluatorWait)
-				modelSrv, err := service.New(context.Background(), model, tfService, fs, metrics, datastores, serviceOpts...)
+				var modelSrv *service.Service
+				var err error
+
+				if model.Platform == "" {
+					// Default to TensorFlow for models without explicit platform
+					model.Platform = "tensorflow"
+				}
+				log.Printf("[%s] Using platform-specific service creation for platform: %s", model.ID, model.Platform)
+				modelSrv, err = service.NewWithPlatform(context.Background(), model, fs, metrics, datastores, sema, cfge.MaxEvaluatorWait, serviceOpts...)
 
 				if err != nil {
 					return fmt.Errorf("failed to create service for model:%v, err:%w", model.ID, err)
@@ -139,7 +156,7 @@ func Build(mux *http.ServeMux, config *Config, datastores map[string]*datastore.
 				lock.Unlock()
 			}
 
-			log.Printf("[%s] model loaded (%s)", model.ID, time.Now().Sub(mstart))
+			log.Printf("[%s] model loaded (%s)", model.ID, time.Since(mstart))
 		}(m)
 	}
 

service/platform/factory.go

@@ -0,0 +1,42 @@
+package platform
+
+import (
+	"fmt"
+	"time"
+
+	"github.com/viant/afs"
+	"github.com/viant/gmetric"
+	"github.com/viant/mly/service/config"
+	"github.com/viant/mly/service/tfmodel"
+	"golang.org/x/sync/semaphore"
+)
+
+// CreateEvaluator creates the appropriate platform evaluator based on the model configuration
+func CreateEvaluator(cfg *config.Model, fs afs.Service, metrics *gmetric.Service, sema *semaphore.Weighted, maxEvaluatorWait time.Duration) (PlatformEvaluator, ModelPlatform, error) {
+	platform := cfg.GetPlatform()
+
+	switch platform {
+	case "tensorflow":
+		// Create TensorFlow service using existing code
+		tfService := tfmodel.NewService(cfg, fs, metrics, sema, maxEvaluatorWait)
+		evaluator := NewTensorFlowEvaluator(tfService)
+		return evaluator, PlatformTensorFlow, nil
+
+	case "triton":
+		evaluator := NewTritonEvaluator(cfg)
+		return evaluator, PlatformTriton, nil
+
+	default:
+		return nil, "", fmt.Errorf("unsupported platform: %s for model %s", platform, cfg.ID)
+	}
+}
+
+// CreateEvaluatorContext creates a platform evaluator with context
+func CreateEvaluatorContext(cfg *config.Model, fs afs.Service, metrics *gmetric.Service, sema *semaphore.Weighted, maxEvaluatorWait time.Duration) (*PlatformEvaluatorContext, error) {
+	evaluator, platform, err := CreateEvaluator(cfg, fs, metrics, sema, maxEvaluatorWait)
+	if err != nil {
+		return nil, err
+	}
+
+	return NewEvaluatorContext(cfg, evaluator, platform), nil
+}