[FSTORE_612] Support for feature monitoring (#1692)

Co-authored-by: Robin Andersson @robzor92
Co-authored-by: Victor Jouffrey @vatj
Co-authored-by: Kenneth Mak @kennethmhc
Co-authored-by: Dhananjay Mukhedkar @dhananjay-mk
Co-authored-by: Ehsan Heydari @ehsan-github

hopsworks-IT/src/test/ruby/spec/feature_group_alert_spec.rb → hopsworks-IT/src/test/ruby/spec/feature_store_alert_spec.rb

@@ -1,21 +1,32 @@
-=begin
- This file is part of Hopsworks
- Copyright (C) 2021, Logical Clocks AB. All rights reserved
+# This file is part of Hopsworks
+# Copyright (C) 2024, Hopsworks AB. All rights reserved
+#
+# Hopsworks is free software: you can redistribute it and/or modify it under the terms of
+# the GNU Affero General Public License as published by the Free Software Foundation,
+# either version 3 of the License, or (at your option) any later version.
+#
+# Hopsworks is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+# without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+# PURPOSE. See the GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License along with this program.
+# If not, see <https://www.gnu.org/licenses/>.
+#
 
- Hopsworks is free software: you can redistribute it and/or modify it under the terms of
- the GNU Affero General Public License as published by the Free Software Foundation,
- either version 3 of the License, or (at your option) any later version.
+describe "On #{ENV['OS']}" do
 
- Hopsworks is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
- without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
- PURPOSE. See the GNU Affero General Public License for more details.
+ before :all do
+ # ensure feature monitoring is enabled
+ @enable_feature_monitoring = getVar('enable_feature_monitoring')
+ setVar('enable_feature_monitoring', "true")
+ end
 
- You should have received a copy of the GNU Affero General Public License along with this program.
- If not, see <https://www.gnu.org/licenses/>.
-=end
+ after :all do
+ # revert feature monitoring flag
+ setVar('enable_feature_monitoring', @enable_feature_monitoring[:value])
+ clean_all_test_projects(spec: "fg_alert")
+ end
 
-describe "On #{ENV['OS']}" do
- after(:all) {clean_all_test_projects(spec: "fg_alert")}
  describe 'Alert' do
  context 'without authentication' do
  before :all do
@@ -31,17 +42,25 @@
  create_fg_alert(@project, @featuregroup, get_fg_alert_success(@project))
  expect_status_details(401)
  end
+ it "should fail to create feature monitoring status" do
+ create_fg_alert(@project, @featuregroup, get_fm_alert_success(@project))
+ expect_status_details(401)
+ end
  end
  context 'with authentication' do
  before :all do
  with_valid_project
  @featuregroup = with_valid_fg(@project)
+ json_result = create_feature_view_from_feature_group(@project.id, get_featurestore_id(@project.id), @featuregroup)
+ @feature_view = JSON.parse(json_result)
  create_fg_alerts(@project, @featuregroup)
+ create_fm_alerts(@project, @featuregroup, @feature_view)
+
  end
  it "should get" do
  get_fg_alerts(@project, @featuregroup)
  expect_status_details(200)
- expect(json_body[:count]).to eq(2)
+ expect(json_body[:count]).to eq(3)
  end
  it "should update" do
  get_fg_alerts(@project, @featuregroup)
@@ -68,7 +87,7 @@
  expect_status_details(201)
  check_route_created(@project, alert[:receiver], alert[:status], fg: @featuregroup)
  get_fg_alerts(@project, @featuregroup)
- expect(json_body[:count]).to eq(3)
+ expect(json_body[:count]).to eq(4)
  end
  it "should fail to create duplicate" do
  create_fg_alert(@project, @featuregroup, get_fg_alert_warning(@project))
@@ -81,7 +100,48 @@
  expect_status_details(204)
  check_route_deleted(@project, alert[:receiver], alert[:status], fg: @featuregroup)
  get_fg_alerts(@project, @featuregroup)
- expect(json_body[:count]).to eq(2)
+ expect(json_body[:count]).to eq(3)
+ end
+ it "should create alert with feature monitoring status" do
+ alert_data = get_fm_alert_success(@project)
+ json_result = create_fg_alert(@project, @featuregroup, alert_data)
+ parsed_alert_json = JSON.parse(json_result)
+ expect_status_details(201)
+ check_route_created_fm(@project, alert_data[:receiver], alert_data[:status], fg: @featuregroup)
+ expect(parsed_alert_json['status']).to eql(alert_data[:status])
+ expect(parsed_alert_json['receiver']).to eql(alert_data[:receiver])
+ expect(parsed_alert_json['severity']).to eql(alert_data[:severity])
+ expect(parsed_alert_json['featureGroupId']).to eql(@featuregroup["id"])
+ end
+ it "should create feature view monitoring alert" do
+ alert_data = get_fm_alert_success(@project)
+ json_result = create_feature_view_alert(@project, @feature_view, alert_data)
+ parsed_alert_json = JSON.parse(json_result)
+ expect_status_details(201)
+ expect(parsed_alert_json['status']).to eql(alert_data[:status])
+ expect(parsed_alert_json['receiver']).to eql(alert_data[:receiver])
+ expect(parsed_alert_json['severity']).to eql(alert_data[:severity])
+ expect(parsed_alert_json['featureViewName']).to eql(@feature_view["name"])
+ expect(parsed_alert_json['featureViewVersion']).to eql(@feature_view["version"])
+ check_route_created_fm(@project, alert_data[:receiver], alert_data[:status], fv: @feature_view)
+ end
+ it "should get and update feature view monitoring alert" do
+ get_featureview_alerts(@project, @feature_view)
+ expect_status_details(200)
+ alert = json_body[:items].detect { |a| a[:status] == "FEATURE_MONITOR_SHIFT_UNDETECTED" && a[:featureViewId]
+ .present? }
+ receiver_original = alert[:receiver]
+ alert[:receiver] = "#{@project[:projectname]}__slack1"
+ json_result = update_featureview_alert(@project, @feature_view, alert[:id], alert)
+ expect_status_details(200)
+ parsed_updated_alert = JSON.parse(json_result)
+ expect(parsed_updated_alert['status']).to eql(alert[:status])
+ expect(parsed_updated_alert['receiver']).to eql(alert[:receiver])
+ expect(parsed_updated_alert['severity']).to eql(alert[:severity])
+ expect(parsed_updated_alert['featureViewName']).to eql(@feature_view["name"])
+ expect(parsed_updated_alert['featureViewVersion']).to eql(@feature_view["version"])
+ check_route_created_fm(@project, alert[:receiver], alert[:status], fv: @feature_view)
+ check_route_deleted_fm(@project, receiver_original, alert[:status], fv: @feature_view)
  end
  it "should cleanup receivers and routes when deleting project" do
  delete_project(@project)
@@ -102,7 +162,7 @@
  featuregroup = with_valid_fg(project)
  create_fg_alerts_global(project, featuregroup)
  get_fg_alerts(project, featuregroup)
- expect(json_body[:count]).to eq(3)
+ expect(json_body[:count]).to eq(5)
  alert_receiver = AlertReceiver.where("name LIKE '#{project[:projectname]}__%'")
  expect(alert_receiver.length()).to eq(0)
  get_routes_admin()
@@ -161,4 +221,4 @@
  end
  end
  end
-end
+end

hopsworks-IT/src/test/ruby/spec/featurestore_statistics_spec.rb

@@ -46,6 +46,14 @@
  commit_metadata = {commitDateString:20201025231125,commitTime:1603667485000,rowsInserted:4,rowsUpdated:0,rowsDeleted:0}
  json_result = commit_cached_featuregroup(@project[:id], @featurestore_id, @stream_feature_group["id"], commit_metadata: commit_metadata)
  expect_status_details(200)
+ # feature view - cached feature group - no time travel
+ json_result = create_feature_view_from_feature_group(@project[:id], @featurestore_id, @cached_feature_group)
+ expect_status_details(201)
+ @cached_feature_view = JSON.parse(json_result)
+ # feature view - stream feature group - time travel
+ json_result = create_feature_view_from_feature_group(@project[:id], @featurestore_id, @stream_feature_group)
+ expect_status_details(201)
+ @stream_feature_view = JSON.parse(json_result)
  # training datasets - with and without splits
  all_metadata = create_featureview_training_dataset_from_project(@project)
  @training_dataset = all_metadata["response"]
@@ -99,6 +107,36 @@
  # all query parameter combinations (including window times) are covered in the unit tests
  end
 
+ # feature view - left feature group - no time travel (for feature monitoring)
+
+ it "should be able to add statistics as a commit to a feature view with a left feature group with time travel disabled (feature monitoring)" do
+ create_statistics_commit_fv(@project[:id], @featurestore_id, @cached_feature_view["name"], @cached_feature_view["version"])
+ expect_status_details(200)
+ end
+
+ it "should fail to add statistics as a commit to a feature view with a left feature group with time travel disabled and window times (feature monitoring)" do
+ create_statistics_commit_fv(@project[:id], @featurestore_id, @cached_feature_view["name"], @cached_feature_view["version"], computation_time: 1597903688010, window_start_commit_time: 1597903688000, window_end_commit_time: 1597903688010)
+ expect_status_details(400, error_code: 270229)
+ # all query parameter combinations (including window times) are covered in the unit tests
+ end
+
+ # feature view - left stream feature group - time travel enable (for feature monitoring)
+
+ it "should be able to add statistics as a commit to a feature view with a left stream feature group (feature monitoring)" do
+ create_statistics_commit_fv(@project[:id], @featurestore_id, @stream_feature_view["name"], @stream_feature_view["version"])
+ expect_status_details(200)
+ end
+
+ it "should be able to add statistics as a commit to a feature view with a left stream feature group with window times (feature monitoring)" do
+ # on two commits
+ create_statistics_commit_fv(@project[:id], @featurestore_id, @stream_feature_view["name"], @stream_feature_view["version"], computation_time: 1603667485000, window_start_commit_time: 1603577485000, window_end_commit_time: 1603667485000)
+ expect_status_details(200)
+ # on a single commit
+ create_statistics_commit_fv(@project[:id], @featurestore_id, @stream_feature_view["name"], @stream_feature_view["version"], computation_time: 1603667485000, window_start_commit_time: 1603667485000, window_end_commit_time: 1603667485000)
+ expect_status_details(200)
+ # all query parameter combinations (including window times) are covered in the unit tests
+ end
+
  # training dataset
 
  it "should be able to add statistics as a commit to a training dataset" do

hopsworks-IT/src/test/ruby/spec/helpers/alert_helper.rb

@@ -299,6 +299,10 @@ def with_receivers(project)
  pagerdutyConfigs: [create_pager_duty_config]))
  create_receivers_checked(project, create_receiver(project, name: "#{project[:projectname]}__slack1",
  slackConfigs: [create_slack_config]))
+ create_receivers_checked(project, create_receiver(project, name: "#{project[:projectname]}__pagerduty1",
+ pagerdutyConfigs: [create_pager_duty_config]))
+ create_receivers_checked(project, create_receiver(project, name: "#{project[:projectname]}__slack2",
+ slackConfigs: [create_slack_config]))
  end
 
  def get_alerts(project, query: "")
@@ -582,6 +586,16 @@ def create_route_match_query(project, receiver, status, job: nil, fg: nil)
  return query
  end
 
+ def create_fm_route_match_query(project, receiver, status, fg:nil, fv: nil)
+ query = "?match=status:#{status}"
+ query = receiver.start_with?("global-receiver__") ?
+ "#{query}&match=type:global-alert-#{receiver.partition('__')[2]}" :
+ "#{query}&match=project:#{project[:projectname]}&match=type:project-alert"
+ query = fv ? "#{query}&match=featureViewName:#{fv['name']}" : query
+ query = fg ? "#{query}&match=featureGroup:#{fg['name']}" : query
+ return query
+ end
+
  def check_route_created(project, receiver, status, job: nil, fg: nil)
  query = create_route_match_query(project, receiver, status, job: job, fg: fg)
  get_routes_by_receiver(project, receiver, query: query)
@@ -603,4 +617,19 @@ def check_route_deleted(project, receiver, status, job: nil, fg: nil)
  expect_status_details(400)
  end
 
-end
+ def check_route_created_fm(project, receiver, status, fg: nil, fv: nil)
+ query = create_fm_route_match_query(project, receiver, status, fg: fg, fv: fv)
+ get_routes_by_receiver(project, receiver, query:query)
+ expect_status_details(200)
+ check_backup_contains_route(json_body)
+ check_alert_receiver_created(receiver)
+ expect(json_body[:receiver]).to eq(receiver)
+ end
+
+ def check_route_deleted_fm(project, receiver, status, fg: nil, fv: nil)
+ query = create_fm_route_match_query(project, receiver, status, fg: fg, fv: fv)
+ get_routes_by_receiver(project, receiver, query: query)
+ expect_status_details(400)
+ end
+
+end

hopsworks-IT/src/test/ruby/spec/helpers/feature_group_alert_helper.rb

@@ -16,11 +16,14 @@
 
 module FeatureGroupAlertHelper
  @@fg_alert_resource = "#{ENV['HOPSWORKS_API']}/project/%{projectId}/featurestores/%{fsId}/featuregroups/%{fgId}/alerts"
-
  @@alert_success = {"status": "SUCCESS", "receiver": "global-receiver__email", "severity": "INFO"}
  @@alert_warning = {"status": "WARNING", "receiver": "global-receiver__slack", "severity": "WARNING"}
  @@alert_failed = {"status": "FAILURE", "receiver": "global-receiver__pagerduty", "severity": "CRITICAL"}
-
+ # feature monitor specific
+ @@featureview_alert_resource = "#{ENV['HOPSWORKS_API']}/project/%{projectId}/featurestores/%{fsId}/featureview/%{name}/version/%{version}/alerts"
+ @@fm_alert_success = {"status": "FEATURE_MONITOR_SHIFT_UNDETECTED", "receiver": "global-receiver__email", "severity":
+ "INFO"}
+ @@fm_alert_failed = {"status": "FEATURE_MONITOR_SHIFT_DETECTED", "receiver": "global-receiver__pagerduty", "severity": "CRITICAL"}
  def get_fg_alert_success(project)
  success = @@alert_success.clone
  success[:receiver] = "#{project[:projectname]}__email"
@@ -80,11 +83,53 @@ def create_fg_alerts_global(project, featuregroup)
  create_fg_alert(project, featuregroup, @@alert_success.clone)
  create_fg_alert(project, featuregroup, @@alert_warning.clone)
  create_fg_alert(project, featuregroup, @@alert_failed.clone)
+ create_fg_alert(project, featuregroup, @@fm_alert_failed.clone)
+ create_fg_alert(project, featuregroup, @@fm_alert_success.clone)
  end
 
  def with_valid_fg(project)
  featurestore_id = get_featurestore_id(project[:id])
  json_result, featuregroup_name = create_cached_featuregroup(project[:id], featurestore_id)
  return JSON.parse(json_result)
  end
-end
+
+ def get_fm_alert_failure(project)
+ failed = @@fm_alert_failed.clone
+ failed[:receiver] = "#{project[:projectname]}__pagerduty"
+ return failed
+ end
+
+ def get_fm_alert_success(project)
+ success = @@fm_alert_success.clone
+ success[:receiver] = "#{project[:projectname]}__email"
+ return success
+ end
+
+ def create_fm_alerts(project, featuregroup, featureview)
+ create_fg_alert(project, featuregroup, get_fm_alert_failure(project))
+ create_feature_view_alert(project, featureview, get_fm_alert_failure(project))
+ end
+
+ def create_feature_view_alert(project, featureview, alert)
+ post "#{@@featureview_alert_resource}" % { projectId: project[:id],
+ fsId: featureview["featurestoreId"],
+ name: featureview["name"],
+ version: featureview["version"]}, alert.to_json
+ end
+
+ def update_featureview_alert(project, featureview, id, alert)
+ put "#{@@featureview_alert_resource}/#{id}" % { projectId: project[:id],
+ fsId: featureview["featurestoreId"],
+ name: featureview["name"],
+ version: featureview["version"]}, alert.to_json
+ end
+
+ def get_featureview_alerts(project, featureview)
+ get "#{@@featureview_alert_resource}" % {projectId: project[:id],
+ fsId: featureview["featurestoreId"],
+ name: featureview["name"],
+ version: featureview["version"]}
+ end
+
+
+end