Skip to content

Commit

Permalink
[HWORKS-919] Create database on the online feature store on-demand (#…
Browse files Browse the repository at this point in the history 
…1458)
  • Loading branch information
@kennethmhc
kennethmhc authored Feb 2, 2024
1 parent 20d009f commit f3791ea
Show file tree
Hide file tree
Showing 14 changed files with 369 additions and 157 deletions.
125 changes: 122 additions & 3 deletions hopsworks-IT/src/test/ruby/spec/featurestore_spec.rb
View file
Original file line number Diff line number Diff line change
Expand Up @@ -112,16 +112,22 @@
end
end
end
describe "ensure currect tables are created in online feature store" do

describe "ensure correct tables are created in online feature store" do
context 'with valid project and online feature store enabled' do
before :all do
if getVar("featurestore_online_enabled") == false
skip "Online Feature Store not enabled, skip online featurestore tests"
end
with_valid_project
# setup online feature store by creating a online feature group (see HWORKS-919)
project = get_project
featurestore_id = get_featurestore_id(project.id)
json_result, featuregroup_name = create_cached_featuregroup(project.id, featurestore_id, online:true)
parsed_json = JSON.parse(json_result)
expect_status_details(201)
end

it "should make sure that kafka_offsets table is created in featurestores" do
project = get_project
tables = Tables.where(TABLE_SCHEMA:project[:projectname], TABLE_NAME:"kafka_offsets")
Expand All @@ -137,6 +143,12 @@
skip "Online Feature Store not enabled, skip online featurestore tests"
end
with_valid_project
# setup online feature store by creating a online feature group (see HWORKS-919)
project = get_project
featurestore_id = get_featurestore_id(project.id)
json_result, featuregroup_name = create_cached_featuregroup(project.id, featurestore_id, online:true)
parsed_json = JSON.parse(json_result)
expect_status_details(201)
end

it "should grant all privileges to the project owner" do
Expand Down Expand Up @@ -211,6 +223,11 @@
update_project({projectId: no_fs_project[:id],
projectName: no_fs_project[:projectname],
services: ["FEATURESTORE"]})
# setup online feature store by creating a online feature group (see HWORKS-919)
featurestore_id = get_featurestore_id(no_fs_project.id)
json_result, featuregroup_name = create_cached_featuregroup(no_fs_project.id, featurestore_id, online:true)
parsed_json = JSON.parse(json_result)
expect_status_details(201)

# Project owner should have the correct permissions on the online feature store
# online fs username are capped to 30 chars
Expand All @@ -234,6 +251,16 @@
it "should not remove users from other projects - see HOPSWORKS-2856" do
demo_project = create_project(projectName = "demo")
demo_demo_project = create_project(projectName = "demo_demo")
# setup online feature store by creating a online feature group (see HWORKS-919)
featurestore_id = get_featurestore_id(demo_project.id)
json_result, featuregroup_name = create_cached_featuregroup(demo_project.id, featurestore_id, online:true)
parsed_json = JSON.parse(json_result)
expect_status_details(201)
# setup online feature store by creating a online feature group (see HWORKS-919)
featurestore_id = get_featurestore_id(demo_demo_project.id)
json_result, featuregroup_name = create_cached_featuregroup(demo_demo_project.id, featurestore_id, online:true)
parsed_json = JSON.parse(json_result)
expect_status_details(201)

delete_project(demo_project)

Expand All @@ -252,7 +279,18 @@
skip "Online Feature Store not enabled, skip online featurestore tests"
end
with_valid_project
# setup online feature store by creating a online feature group (see HWORKS-919)
project = get_project
featurestore_id = get_featurestore_id(project.id)
json_result, featuregroup_name = create_cached_featuregroup(project.id, featurestore_id, online:true)
parsed_json = JSON.parse(json_result)
expect_status_details(201)
@shared_project = create_project
# setup online feature store by creating a online feature group (see HWORKS-919)
featurestore_id = get_featurestore_id(@shared_project.id)
json_result, featuregroup_name = create_cached_featuregroup(@shared_project.id, featurestore_id, online:true)
parsed_json = JSON.parse(json_result)
expect_status_details(201)
@shared_user_do = create_user
@shared_user_ds = create_user
add_member_to_project(@shared_project, @shared_user_do[:email], 'Data owner')
Expand Down Expand Up @@ -320,5 +358,86 @@
end
end
end

describe "check permission when shared online feature store has not been initialised" do
before :all do
if getVar("featurestore_online_enabled") == false
skip "Online Feature Store not enabled, skip online featurestore tests"
end
@project_a = create_project
@user_a_do = create_user
@user_a_ds = create_user
add_member_to_project(@project_a, @user_a_do[:email], 'Data owner')
add_member_to_project(@project_a, @user_a_ds[:email], 'Data scientist')
# setup online feature store by creating a online feature group (see HWORKS-919)
# before adding member
@project_b = create_project
featurestore_id = get_featurestore_id(@project_b.id)
json_result, featuregroup_name = create_cached_featuregroup(@project_b.id, featurestore_id, online:true)
parsed_json = JSON.parse(json_result)
@user_b_do = create_user
@user_b_ds = create_user
add_member_to_project(@project_b, @user_b_do[:email], 'Data owner')
add_member_to_project(@project_b, @user_b_ds[:email], 'Data scientist')
@project_c = create_project
@user_c_do = create_user
@user_c_ds = create_user
add_member_to_project(@project_c, @user_c_do[:email], 'Data owner')
add_member_to_project(@project_c, @user_c_ds[:email], 'Data scientist')
# setup online feature store by creating a online feature group (see HWORKS-919)
# after adding member
featurestore_id = get_featurestore_id(@project_c.id)
json_result, featuregroup_name = create_cached_featuregroup(@project_c.id, featurestore_id, online:true)
parsed_json = JSON.parse(json_result)
end

it "Project C should be able to share with project A (has not been initialised)" do
featurestore = "#{@project_c['projectname'].downcase}_featurestore.db"
# Project C should be able to share with project A without error,
# but privileges will not be added to information_schema.SCHEMA_PRIVILEGES yet.
share_dataset(@project_c, featurestore, @project_a['projectname'], datasetType: "&type=FEATURESTORE")
accept_dataset(@project_a, "#{@project_c['projectname']}::#{featurestore}",
datasetType: "&type=FEATURESTORE")
end

it "Project A (has not been initialised) should be able to share with project B (A )" do
featurestore = "#{@project_a['projectname'].downcase}_featurestore.db"
# Project A should be able to share with project B without error
# but privileges will not be added to information_schema.SCHEMA_PRIVILEGES yet.
share_dataset(@project_a, featurestore, @project_b['projectname'], datasetType: "&type=FEATURESTORE")
accept_dataset(@project_b, "#{@project_a['projectname']}::#{featurestore}",
datasetType: "&type=FEATURESTORE")
end

it "After project A has initialised online feature store, privileges should be set properly" do
# Project B should have access to project A, and project A should have access to project C
# setup online feature store in project A by creating a online feature group
featurestore_id = get_featurestore_id(@project_a.id)
json_result, featuregroup_name = create_cached_featuregroup(@project_a.id, featurestore_id, online:true)
parsed_json = JSON.parse(json_result)

# make sure project B get access to project A
online_db_name_ds = "#{@project_b[:projectname]}_#{@user_b_ds[:username]}"[0..30]
grantee_ds = "'#{online_db_name_ds}'@'%'"
online_db_name_do = "#{@project_b[:projectname]}_#{@user_b_do[:username]}"[0..30]
grantee_do = "'#{online_db_name_do}'@'%'"
privileges_ds = SchemaPrivileges.where(TABLE_SCHEMA:@project_a[:projectname], GRANTEE:grantee_ds)
privileges_do = SchemaPrivileges.where(TABLE_SCHEMA:@project_a[:projectname], GRANTEE:grantee_do)
expect(privileges_ds.length).to eq(1)
expect(privileges_do.length).to eq(1)

# make sure project A cannot access to project C
online_db_name_ds = "#{@project_a[:projectname]}_#{@user_a_ds[:username]}"[0..30]
grantee_ds = "'#{online_db_name_ds}'@'%'"
online_db_name_do = "#{@project_a[:projectname]}_#{@user_a_do[:username]}"[0..30]
grantee_do = "'#{online_db_name_do}'@'%'"

privileges_ds = SchemaPrivileges.where(TABLE_SCHEMA:@project_c[:projectname], GRANTEE:grantee_ds)
privileges_do = SchemaPrivileges.where(TABLE_SCHEMA:@project_c[:projectname], GRANTEE:grantee_do)

expect(privileges_ds.length).to eq(1)
expect(privileges_do.length).to eq(1)
end
end
end
end
147 changes: 82 additions & 65 deletions hopsworks-IT/src/test/ruby/spec/storage_connector_spec.rb
View file
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,89 @@

describe "Create, delete and update operations on storage connectors in a specific featurestore" do

describe 'with valid project, featurestore service enabled' do
describe "online feature store storage connector" do
context "with valid project, featurestore service enabled, and with storage connector flags enabled" do
before :all do
with_valid_project
# setup online feature store by creating a online feature group (see HWORKS-919)
project = get_project
featurestore_id = get_featurestore_id(project.id)
json_result, featuregroup_name = create_cached_featuregroup(project.id, featurestore_id, online:true)
parsed_json = JSON.parse(json_result)
expect_status_details(201)
end

after :each do
create_session(@project[:username], "Pass123")
end

it "online storage connector connection string should contain the IP of the mysql" do
# Storage connector looks like this: [project name]_[username]_onlinefeaturestore
connector_name = "#{@project['projectname']}_#{@user['username']}_onlinefeaturestore"
featurestore_id = get_featurestore_id(@project['id'])
connector_json = get_storage_connector(@project['id'], featurestore_id, connector_name)
connector = JSON.parse(connector_json)

expect(connector['connectionString']).to match(/jdbc:mysql:\/\/\d{1,3}\.\d{1,3}\.\d{1,3}.\d{1,3}/)
end

it "online storage connector connection string should be stored with consul name in the database" do
# Storage connector looks like this: [project name]_[username]_onlinefeaturestore
connector_name = "#{@project['projectname']}_#{@user['username']}_onlinefeaturestore"
connector_db = FeatureStoreConnector.find_by(name: connector_name)
jdbc_connector_db = FeatureStoreJDBCConnector.find_by(id: connector_db.jdbc_id)

expect(jdbc_connector_db[:connection_string]).to start_with("jdbc:mysql://onlinefs.mysql.service.consul")
end

it "online storage connector should contain the driver class and isolationLevel" do
connector_name = "#{@project['projectname']}_#{@user['username']}_onlinefeaturestore"
featurestore_id = get_featurestore_id(@project['id'])
connector_json = get_storage_connector(@project['id'], featurestore_id, connector_name)
connector = JSON.parse(connector_json)

arguments_hash = connector['arguments']
puts arguments_hash
expect(arguments_hash.find{ |item| item['name'] == 'driver' }['value']).to eql("com.mysql.cj.jdbc.Driver")
expect(arguments_hash.find{ |item| item['name'] == 'isolationLevel' }['value']).to eql("NONE")
end

it "should get online storage connector from base project when accessing a shared feature store" do
project = get_project
base_featurestore_id = get_featurestore_id(project.id)
reset_session

context "with storage connector flags enabled" do
#create another project
projectname = "project_#{short_random_id}"
shared_fs_project = create_project_by_name(projectname)
shared_fs_id = get_featurestore_id(shared_fs_project.id)
# login with user for project and share dataset
create_session(shared_fs_project[:username], "Pass123")
featurestore = "#{shared_fs_project[:projectname].downcase}_featurestore.db"
share_dataset(shared_fs_project, featurestore, project[:projectname], datasetType: "&type=FEATURESTORE")
reset_session
#login with user for shared_fs_project and accept dataset
create_session(project[:username],"Pass123")
accept_dataset(project, "#{shared_fs_project[:projectname]}::#{featurestore}", datasetType: "&type=FEATURESTORE")

base_project_connector = "#{project['projectname']}_#{@user['username']}_onlinefeaturestore"
connector_name = "onlinefeaturestore"

#connector from shared fs should use base project connector
connector_json = get_storage_connector(project['id'], shared_fs_id, connector_name)
connector = JSON.parse(connector_json)
expect(connector["name"]).to eql(base_project_connector)

#connector from base should use base project connector
connector_json = get_storage_connector(project['id'], base_featurestore_id, connector_name)
connector = JSON.parse(connector_json)
expect(connector["name"]).to eql(base_project_connector)
end
end
end

describe "other storage connectors" do
context "with valid project, featurestore service enabled, and with storage connector flags enabled" do
before :all do
with_valid_project
end
Expand Down Expand Up @@ -454,69 +534,6 @@
expect(parsed_json2["storageConnectorType"] == "JDBC").to be true
expect(parsed_json2["connectionString"] == "jdbc://test3").to be true
end

it "online storage connector connection string should contain the IP of the mysql" do
# Storage connector looks like this: [project name]_[username]_onlinefeaturestore
connector_name = "#{@project['projectname']}_#{@user['username']}_onlinefeaturestore"
featurestore_id = get_featurestore_id(@project['id'])
connector_json = get_storage_connector(@project['id'], featurestore_id, connector_name)
connector = JSON.parse(connector_json)

expect(connector['connectionString']).to match(/jdbc:mysql:\/\/\d{1,3}\.\d{1,3}\.\d{1,3}.\d{1,3}/)
end

it "online storage connector connection string should be stored with consul name in the database" do
# Storage connector looks like this: [project name]_[username]_onlinefeaturestore
connector_name = "#{@project['projectname']}_#{@user['username']}_onlinefeaturestore"
connector_db = FeatureStoreConnector.find_by(name: connector_name)
jdbc_connector_db = FeatureStoreJDBCConnector.find_by(id: connector_db.jdbc_id)

expect(jdbc_connector_db[:connection_string]).to start_with("jdbc:mysql://onlinefs.mysql.service.consul")
end

it "online storage connector should contain the driver class and isolationLevel" do
connector_name = "#{@project['projectname']}_#{@user['username']}_onlinefeaturestore"
featurestore_id = get_featurestore_id(@project['id'])
connector_json = get_storage_connector(@project['id'], featurestore_id, connector_name)
connector = JSON.parse(connector_json)

arguments_hash = connector['arguments']
puts arguments_hash
expect(arguments_hash.find{ |item| item['name'] == 'driver' }['value']).to eql("com.mysql.cj.jdbc.Driver")
expect(arguments_hash.find{ |item| item['name'] == 'isolationLevel' }['value']).to eql("NONE")
end

it "should get online storage connector from base project when accessing a shared feature store" do
project = get_project
base_featurestore_id = get_featurestore_id(project.id)
reset_session

#create another project
projectname = "project_#{short_random_id}"
shared_fs_project = create_project_by_name(projectname)
shared_fs_id = get_featurestore_id(shared_fs_project.id)
# login with user for project and share dataset
create_session(shared_fs_project[:username], "Pass123")
featurestore = "#{shared_fs_project[:projectname].downcase}_featurestore.db"
share_dataset(shared_fs_project, featurestore, project[:projectname], datasetType: "&type=FEATURESTORE")
reset_session
#login with user for shared_fs_project and accept dataset
create_session(project[:username],"Pass123")
accept_dataset(project, "#{shared_fs_project[:projectname]}::#{featurestore}", datasetType: "&type=FEATURESTORE")

base_project_connector = "#{project['projectname']}_#{@user['username']}_onlinefeaturestore"
connector_name = "onlinefeaturestore"

#connector from shared fs should use base project connector
connector_json = get_storage_connector(project['id'], shared_fs_id, connector_name)
connector = JSON.parse(connector_json)
expect(connector["name"]).to eql(base_project_connector)

#connector from base should use base project connector
connector_json = get_storage_connector(project['id'], base_featurestore_id, connector_name)
connector = JSON.parse(connector_json)
expect(connector["name"]).to eql(base_project_connector)
end
end

context "with storage connector flags disabled" do
Expand Down
7 changes: 1 addition & 6 deletions ...hops/hopsworks/api/featurestore/storageconnector/FeaturestoreStorageConnectorService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -283,12 +283,7 @@ public Response getOnlineFeaturestoreStorageConnector(@Context SecurityContext s
throw new FeaturestoreException(RESTCodes.FeaturestoreErrorCode.FEATURESTORE_ONLINE_NOT_ENABLED,
Level.FINE, "Online Featurestore is not enabled for this Hopsworks cluster.");
}
if (!onlineFeaturestoreController.checkIfDatabaseExists(
onlineFeaturestoreController.getOnlineFeaturestoreDbName(featurestore.getProject()))) {
throw new FeaturestoreException(RESTCodes.FeaturestoreErrorCode.FEATURESTORE_ONLINE_NOT_ENABLED,
Level.FINE, "Online Featurestore is not enabled for this project. To enable online feature store," +
" talk to an administrator.");
}
// continue even if featureStoreDb does not exist, see HWORKS-919
Users user = jWTHelper.getUserPrincipal(sc);
FeaturestoreStorageConnectorDTO featurestoreJdbcConnectorDTO =
storageConnectorController.getOnlineFeaturestoreConnector(user, project);
Expand Down
1 change: 1 addition & 0 deletions ...ommon/src/main/java/io/hops/hopsworks/common/dao/user/security/secrets/SecretsFacade.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ public Secret findById(SecretId id) {

public void persist(Secret secret) {
entityManager.persist(secret);
entityManager.flush();
}

public void update(Secret secret) {
Expand Down
Loading

0 comments on commit f3791ea

Please sign in to comment.