feat(api): allow enable mlobs from API directly (#633)

# Description
- deprecate `Model.ObservabilitySupported`
- use `VersionEndpoint.ModelObservability.Enabled` as SSOT for MLObs
configuration

# Modifications
- remove the use of `Model.ObservabilitySupported`
- use `VersionEndpoint.EnableModelObservability` value as
`VersionEndpoint.ModelObservability.Enabled` if
`VersionEndpoint.ModelObservability` is not set
- remove unit test with `Model.ObservabilitySupported` logic
- adjust unit test mock call

# Checklist
- [x] Added PR label
- [ ] Added unit test, integration, and/or e2e tests
- [x] Tested locally
- [ ] Updated documentation
- [ ] Update Swagger spec if the PR introduce API changes
- [ ] Regenerated Golang and Python client if the PR introduces API
changes

---------

Co-authored-by: Muhammad Naufal Andika Natsir Putra <[email protected]>

Author: naufalandika

api/api/validator.go

@@ -179,13 +179,11 @@ func modelObservabilityValidation(endpoint *models.VersionEndpoint, model *model
 		if !endpoint.IsModelMonitoringEnabled() {
 			return nil
 		}
+
 		if !slices.Contains(supportedObservabilityModelTypes, model.Type) {
 			return fmt.Errorf("%s: %w", model.Type, ErrUnsupportedObservabilityModelType)
 		}
 
-		if !model.ObservabilitySupported {
-			return fmt.Errorf("model observability is not supported for this model")
-		}
 		return nil
 	})
 }

api/api/version_endpoints_api.go

@@ -154,6 +154,7 @@ func (c *EndpointsController) CreateEndpoint(r *http.Request, vars map[string]st
 		}
 
 		newEndpoint.EnvironmentName = env.Name
+		newEndpoint.SetModeObservabilityIfNil()
 	}
 
 	validationRules := []requestValidator{
@@ -210,6 +211,8 @@ func (c *EndpointsController) UpdateEndpoint(r *http.Request, vars map[string]st
 		return BadRequest("Unable to parse body as version endpoint resource")
 	}
 
+	newEndpoint.SetModeObservabilityIfNil()
+
 	env, err := c.AppContext.EnvironmentService.GetEnvironment(newEndpoint.EnvironmentName)
 	if err != nil {
 		if errors.Is(err, gorm.ErrRecordNotFound) {

api/api/version_endpoints_api_test.go

@@ -1068,7 +1068,9 @@ func TestCreateEndpoint(t *testing.T) {
 						Value: "1",
 					},
 				}),
-				EnableModelObservability: true,
+				ModelObservability: &models.ModelObservability{
+					Enabled: true,
+				},
 			},
 			modelService: func() *mocks.ModelsService {
 				svc := &mocks.ModelsService{}
@@ -1200,105 +1202,6 @@ func TestCreateEndpoint(t *testing.T) {
 				},
 			},
 		},
-		{
-			desc: "Fail when try to enable model observability but the model is not supported yet",
-			vars: map[string]string{
-				"model_id":   "1",
-				"version_id": "1",
-			},
-			requestBody: &models.VersionEndpoint{
-				ID:              uuid,
-				VersionID:       models.ID(1),
-				VersionModelID:  models.ID(1),
-				ServiceName:     "sample",
-				Namespace:       "sample",
-				EnvironmentName: "dev",
-				Message:         "",
-				ResourceRequest: &models.ResourceRequest{
-					MinReplica:    1,
-					MaxReplica:    4,
-					CPURequest:    resource.MustParse("1"),
-					MemoryRequest: resource.MustParse("1Gi"),
-				},
-				EnvVars: models.EnvVars([]models.EnvVar{
-					{
-						Name:  "WORKER",
-						Value: "1",
-					},
-				}),
-				EnableModelObservability: true,
-			},
-			modelService: func() *mocks.ModelsService {
-				svc := &mocks.ModelsService{}
-				svc.On("FindByID", mock.Anything, models.ID(1)).Return(&models.Model{
-					ID:                     models.ID(1),
-					Name:                   "model-1",
-					ProjectID:              models.ID(1),
-					Project:                mlp.Project{},
-					ExperimentID:           1,
-					Type:                   "pyfunc",
-					MlflowURL:              "",
-					Endpoints:              nil,
-					ObservabilitySupported: false,
-				}, nil)
-				return svc
-			},
-			versionService: func() *mocks.VersionsService {
-				svc := &mocks.VersionsService{}
-				svc.On("FindByID", mock.Anything, models.ID(1), models.ID(1), mock.Anything).Return(&models.Version{
-					ID:      models.ID(1),
-					ModelID: models.ID(1),
-					Model: &models.Model{
-						ID:           models.ID(1),
-						Name:         "model-1",
-						ProjectID:    models.ID(1),
-						Project:      mlp.Project{},
-						ExperimentID: 1,
-						Type:         "pyfunc",
-						MlflowURL:    "",
-						Endpoints:    nil,
-					},
-				}, nil)
-				return svc
-			},
-			envService: func() *mocks.EnvironmentService {
-				svc := &mocks.EnvironmentService{}
-				svc.On("GetDefaultEnvironment").Return(&models.Environment{
-					ID:         models.ID(1),
-					Name:       "dev",
-					Cluster:    "dev",
-					IsDefault:  &trueBoolean,
-					Region:     "id",
-					GcpProject: "dev-proj",
-					MaxCPU:     "1",
-					MaxMemory:  "1Gi",
-				}, nil)
-				svc.On("GetEnvironment", "dev").Return(&models.Environment{
-					ID:         models.ID(1),
-					Name:       "dev",
-					Cluster:    "dev",
-					IsDefault:  &trueBoolean,
-					Region:     "id",
-					GcpProject: "dev-proj",
-					MaxCPU:     "1",
-					MaxMemory:  "1Gi",
-				}, nil)
-				return svc
-			},
-			endpointService: func() *mocks.EndpointsService {
-				svc := &mocks.EndpointsService{}
-				svc.On("CountEndpoints", context.Background(), mock.Anything, mock.Anything).Return(0, nil)
-				return svc
-			},
-			monitoringConfig: config.MonitoringConfig{},
-			feastCoreMock: func() *feastmocks.CoreServiceClient {
-				return &feastmocks.CoreServiceClient{}
-			},
-			expected: &Response{
-				code: http.StatusBadRequest,
-				data: Error{Message: "Request validation failed: model observability is not supported for this model"},
-			},
-		},
 		{
 			desc: "Should return 400 if UPI is not supported",
 			vars: map[string]string{
@@ -4004,7 +3907,9 @@ func TestUpdateEndpoint(t *testing.T) {
 						Value: "1",
 					},
 				}),
-				EnableModelObservability: true,
+				ModelObservability: &models.ModelObservability{
+					Enabled: true,
+				},
 			},
 			modelService: func() *mocks.ModelsService {
 				svc := &mocks.ModelsService{}
@@ -4083,7 +3988,9 @@ func TestUpdateEndpoint(t *testing.T) {
 							Value: "1",
 						},
 					}),
-					EnableModelObservability: false,
+					ModelObservability: &models.ModelObservability{
+						Enabled: false,
+					},
 				}, nil)
 				svc.On("DeployEndpoint", context.Background(), mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&models.VersionEndpoint{
 					ID:                   uuid,
@@ -4114,8 +4021,10 @@ func TestUpdateEndpoint(t *testing.T) {
 							Value: "1",
 						},
 					}),
-					CreatedUpdated:           models.CreatedUpdated{},
-					EnableModelObservability: true,
+					CreatedUpdated: models.CreatedUpdated{},
+					ModelObservability: &models.ModelObservability{
+						Enabled: true,
+					},
 				}, nil)
 				return svc
 			},
@@ -4150,8 +4059,10 @@ func TestUpdateEndpoint(t *testing.T) {
 							Value: "1",
 						},
 					}),
-					CreatedUpdated:           models.CreatedUpdated{},
-					EnableModelObservability: true,
+					CreatedUpdated: models.CreatedUpdated{},
+					ModelObservability: &models.ModelObservability{
+						Enabled: true,
+					},
 				},
 			},
 		},
@@ -4696,7 +4607,9 @@ func TestUpdateEndpoint(t *testing.T) {
 						Value: "1",
 					},
 				}),
-				EnableModelObservability: true,
+				ModelObservability: &models.ModelObservability{
+					Enabled: true,
+				},
 			},
 			modelService: func() *mocks.ModelsService {
 				svc := &mocks.ModelsService{}
@@ -4775,7 +4688,9 @@ func TestUpdateEndpoint(t *testing.T) {
 							Value: "1",
 						},
 					}),
-					EnableModelObservability: false,
+					ModelObservability: &models.ModelObservability{
+						Enabled: false,
+					},
 				}, nil)
 				return svc
 			},

api/models/model_endpoint.go

@@ -38,6 +38,10 @@ type ModelEndpoint struct {
 }
 
 func (me *ModelEndpoint) GetVersionEndpoint() *VersionEndpoint {
+	if me == nil {
+		return nil
+	}
+
 	if me.Rule == nil || len(me.Rule.Destination) == 0 {
 		return nil
 	}

api/models/model_observability.go

@@ -16,6 +16,14 @@ type ModelObservability struct {
 	PredictionLogIngestionResourceRequest *WorkerResourceRequest `json:"prediction_log_ingestion_resource_request"`
 }
 
+func (mo *ModelObservability) IsEnabled() bool {
+	if mo == nil {
+		return false
+	}
+
+	return mo.Enabled
+}
+
 // GroundTruthSource represents the source configuration for ground truth data.
 type GroundTruthSource struct {
 	TableURN             string `json:"table_urn"`

api/models/version_endpoint.go

@@ -134,10 +134,10 @@ func (ve *VersionEndpoint) IsServing() bool {
 }
 
 func (ve *VersionEndpoint) IsModelMonitoringEnabled() bool {
-	if ve.ModelObservability == nil {
-		return ve.EnableModelObservability
+	if ve == nil {
+		return false
 	}
-	return ve.ModelObservability.Enabled
+	return ve.ModelObservability.IsEnabled()
 }
 
 func (ve *VersionEndpoint) Hostname() string {
@@ -184,6 +184,23 @@ func (ve *VersionEndpoint) ParsedURL() (*url.URL, error) {
 	return parsedURL, nil
 }
 
+// [TODO]: deprecate this after deprecating VersionEndpoint.EnableModelObservability
+// perviously we only have VersionEndpoint.EnableModelObservability and now we want to deprecate it
+// and only read/write to VersionEndpoint.ModelObservability instead. to allow backward compatibility if the user
+// only set VersionEndpoint.EnableModelObservability but not VersionEndpoint.ModelObservability we will use the
+// VersionEndpoint.EnableModelObservability value as VersionEndpoint.ModelObservability.EnableModelObservability
+func (ve *VersionEndpoint) SetModeObservabilityIfNil() {
+	if ve == nil {
+		return
+	}
+	if ve.ModelObservability != nil {
+		return
+	}
+	ve.ModelObservability = &ModelObservability{
+		Enabled: ve.EnableModelObservability,
+	}
+}
+
 type EndpointMonitoringURLParams struct {
 	Cluster      string
 	Project      string

api/models/version_endpoint_test.go

@@ -244,38 +244,20 @@ func TestVersionEndpoint_IsModelMonitoringEnabled(t *testing.T) {
 		want            bool
 	}{
 		{
-			name: "model observability is nil but enable model observability is true",
-			versionEndpoint: &VersionEndpoint{
-				ModelObservability:       nil,
-				EnableModelObservability: true,
-			},
-			want: true,
-		},
-		{
-			name: "model observability is nil and enable model observability is false",
-			versionEndpoint: &VersionEndpoint{
-				ModelObservability:       nil,
-				EnableModelObservability: false,
-			},
-			want: false,
-		},
-		{
-			name: "model observability is not nil and enabled is true",
+			name: "model observability enabled is true",
 			versionEndpoint: &VersionEndpoint{
 				ModelObservability: &ModelObservability{
 					Enabled: true,
 				},
-				EnableModelObservability: true,
 			},
 			want: true,
 		},
 		{
-			name: "model observability is not nil and enabled is false",
+			name: "model observability enabled is false",
 			versionEndpoint: &VersionEndpoint{
 				ModelObservability: &ModelObservability{
 					Enabled: false,
 				},
-				EnableModelObservability: false,
 			},
 			want: false,
 		},

api/pkg/observability/event/event.go

@@ -34,10 +34,6 @@ func NewEventProducer(jobProducer queue.Producer, observabilityPublisherStorage
 }
 
 func (e *eventProducer) ModelEndpointChangeEvent(modelEndpoint *models.ModelEndpoint, model *models.Model) error {
-	if !model.ObservabilitySupported {
-		return nil
-	}
-
 	ctx := context.Background()
 	publisher, err := e.observabilityPublisherStorage.GetByModelID(ctx, model.ID)
 	if err != nil {
@@ -69,6 +65,10 @@ func (e *eventProducer) ModelEndpointChangeEvent(modelEndpoint *models.ModelEndp
 	}
 
 	versionEndpoint := modelEndpoint.GetVersionEndpoint()
+	if !versionEndpoint.IsModelMonitoringEnabled() {
+		return nil
+	}
+
 	version, err := e.findVersionWithModelSchema(ctx, versionEndpoint.VersionID, model.ID)
 	if err != nil {
 		return err
@@ -93,10 +93,6 @@ func (e *eventProducer) ModelEndpointChangeEvent(modelEndpoint *models.ModelEndp
 }
 
 func (e *eventProducer) VersionEndpointChangeEvent(versionEndpoint *models.VersionEndpoint, model *models.Model) error {
-	if !model.ObservabilitySupported {
-		return nil
-	}
-
 	// check if version endpoint is used by the model endpoint
 	// if version endpoint is not serving skipping deployment
 	if versionEndpoint.Status != models.EndpointServing {