Skip to main content
Version: v2.1.0

System events

Feature Store core service notifies 3rd party systems about your actions by sending events to a pointed Kafka topic or logs.

Enable notifications​

To enable notifications users have to set the notifications.channels parameter to one of the following values:

  • kafka
  • logs
  • kafka,logs

Then, configure the Kafka topic with the proper parameter messaging.kafka.notifications.*.

We recommend providing that configuration through Helm values:

Helm ValueDefaultDescription
global.notifications.channelsemptySet value (e.g. kafka to enable notifications)
global.messaging.kafka.topics.notifications(no default value)Name of notifications topic
note

It is possible to automatically configure new topics by the Feature Store. Please review the global.config.messaging.kafka parameters described in Helm values.

Events producer​

The core service will send an event every time the user makes an API call to the backend. Based on that information, consumer service has knowledge about all actions happening in the Feature Store. That information can be used through an alert system to notify the operator about users' actions.

The following is an example triggered while creating a new project:

project = client.projects.create("test")
{
  "seq": "2022-07-20T08:15:46.861903-62d7b9b2d209b07c1d88f720",
  "method": "CreateProject",
  "timestamp": "2022-07-20T08:15:46.861903",
  "requestId": "fdb98051-c732-49cd-8e5b-8ba48fb325a2",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "62d7b9b2d209b07c1d88f720",
  "projectName": "test1"
}
project = client.projects.get("test")
{
  "seq": "2022-10-03T16:57:58.300710-633af8600423a6148368033e",
  "method": "GetProject",
  "timestamp": "2022-10-03T16:57:58.30071",
  "requestId": "a02d7618-869f-4330-8b91-cc523d2d6ad1",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "62d7b9b2d209b07c1d88f720",
  "projectName": "test"
}
project.delete()
{
  "seq": "2022-10-04T08:35:06.673156-633bd413708da32636a7cdbe",
  "method": "DeleteProject",
  "timestamp": "2022-10-04T08:35:06.673156",
  "requestId": "855981af-c82a-48d3-8b32-828f21e86c27",
  "userId": "8cf44396-1e80-4bd4-8d8d-c5e6befb4f2d",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "62d7b9b2d209b07c1d88f720",
  "projectName": "test"
}
client.projects.list()
{
  "seq": "2022-10-04T08:37:35.166016-8cf44396-1e80-4bd4-8d8d-c5e6befb4f2d",
  "method": "ListProjects",
  "timestamp": "2022-10-04T08:37:35.166016",
  "requestId": "5996c73a-3179-4ad4-aec1-82a215606283",
  "userId": "8cf44396-1e80-4bd4-8d8d-c5e6befb4f2d",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)"
}
project.add_consumers(["dev@h2o.ai"]) # add permissions
{
  "seq": "2022-10-04T08:41:06.803045-633bd481708da32636a7cdc8",
  "method": "AddProjectPermission",
  "timestamp": "2022-10-04T08:41:06.803045",
  "requestId": "a628c9fd-f849-4b91-b764-88e11260ac4f",
  "userId": "8cf44396-1e80-4bd4-8d8d-c5e6befb4f2d",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "633bd481708da32636a7cdc8",
  "projectName": "test3",
  "additionalData": {
    "permissionType": "Consumer",
    "users": "dev@h2o.ai"
  }
}
project.remove_consumers(["dev@h2o.ai"])
{
  "seq": "2022-10-04T08:42:50.347749-633bd481708da32636a7cdc8",
  "method": "RemoveProjectPermission",
  "timestamp": "2022-10-04T08:42:50.347749",
  "requestId": "ef174341-84ce-46e7-a3a7-4eb92e023fa4",
  "userId": "8cf44396-1e80-4bd4-8d8d-c5e6befb4f2d",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "633bd481708da32636a7cdc8",
  "projectName": "test3",
  "additionalData": {
    "permissionType": "Consumer",
    "users": "dev@h2o.ai"
  }
}
project.description = "better description" # similar will be for other fields update
{
  "seq": "2022-10-04T08:45:46.111069-633bd481708da32636a7cdc8",
  "method": "UpdateProjectFields",
  "timestamp": "2022-10-04T08:45:46.111069",
  "requestId": "e6d7d6df-ef0d-4221-9536-dd9fb175ee6c",
  "userId": "8cf44396-1e80-4bd4-8d8d-c5e6befb4f2d",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "633bd481708da32636a7cdc8",
  "projectName": "test3",
  "additionalData": {
    "updatedValue": "better description"
  }
}
fs = project.feature_sets.register(csv_schema, "fs_test")
{
  "seq": "2022-07-20T10:30:04.639687-62d7bd0c93f57739745041a5",
  "sourceRequest": "",
  "method": "OfflineFeatureSetRegister",
  "timestamp": "2022-07-20T10:30:04.639687",
  "requestId": "737ed314-bf29-4d01-aecd-3c4f1ecb21c2",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "62d7bd0c93f57739745041a5",
  "projectName": "test1",
  "featureSetName": "fs_test",
  "additionalData": {
    "partitionBy": "time_travel_column_auto_generated",
    "timeTravelColumn": "",
    "primaryKey": "",
    "createdDateTime": "2022-07-20T08:30:04.446074",
    "owner": "dev@h2o.ai"
  }
}
fs = project.feature_sets.get("fs_test")
{
  "seq": "2022-10-04T09:08:02.048966-633af7b80423a61483680335",
  "sourceRequest": "",
  "method": "GetFeatureSet",
  "timestamp": "2022-10-04T09:08:02.048966",
  "requestId": "0709aeb8-e322-4c22-9bff-b16dbe4dba5b",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "633af7b80423a61483680335",
  "projectName": "test_project",
  "featureSetName": "fs_test",
  "additionalData": {
    "featureSetVersion": "1.1"
  }
}
fs_new = fs.create_new_version(csv_schema,"new version")
{
  "seq": "2022-10-04T09:09:56.742789-633af7b80423a61483680335",
  "sourceRequest": "",
  "method": "CreateNewFeatureSetVersion",
  "timestamp": "2022-10-04T09:09:56.742789",
  "requestId": "0dd81fd4-8f54-48f6-b816-7a81fccdd377",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "633af7b80423a61483680335",
  "projectName": "test_project_123",
  "featureSetName": "test_fs",
  "additionalData": {
    "updatedVersion": "2.0"
  }
}
fs.delete()
{
  "seq": "2022-10-04T09:31:57.332711-633af7b80423a61483680335",
  "sourceRequest": "",
  "method": "DeleteFeatureSet",
  "timestamp": "2022-10-04T09:31:57.332711",
  "requestId": "4f0c0f44-9b39-48aa-8695-7f3beb079645",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "633af7b80423a61483680335",
  "projectName": "test_project_123",
  "featureSetName": "test_fs"
}
fs.description = "test description" # similar will be for other fields update
{
  "seq": "2022-10-04T09:15:55.695864-633af7b80423a61483680335",
  "sourceRequest": "",
  "method": "UpdateFeatureSetFields",
  "timestamp": "2022-10-04T09:15:55.695864",
  "requestId": "fe6a1129-c9d6-4ac9-9442-9b4f39cc35dc",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "633af7b80423a61483680335",
  "projectName": "test_project_123",
  "featureSetName": "test_fs1",
  "additionalData": {
    "description": "test description"
  }
}

{
  "seq": "2022-10-04T09:15:55.924144-633af7b80423a61483680335",
  "sourceRequest": "",
  "method": "GetFeatureSetLastMinor",
  "timestamp": "2022-10-04T09:15:55.924144",
  "requestId": "4bc17a4f-4818-497f-956d-ae5fb75824a2",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "633af7b80423a61483680335",
  "projectName": "test_project_123",
  "featureSetName": "test_fs1",
  "additionalData": {
    "featureSetVersion": "1.2"
  }
}
fs.list_versions()
{
  "seq": "2022-10-04T09:20:08.376487-17a91bc8-3733-4d40-af33-e729cce6823a",
  "method": "ListFeatureSetVersions",
  "timestamp": "2022-10-04T09:20:08.376487",
  "requestId": "c66fe245-e7a8-4c27-9272-63f264cd0e63",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)"
}
project.feature_sets.list()
{
  "seq": "2022-10-03T16:58:40.040700-17a91bc8-3733-4d40-af33-e729cce6823a",
  "method": "ListFeatureSets",
  "timestamp": "2022-10-03T16:58:40.0407",
  "requestId": "1cd48fff-ba6f-4602-bbb8-f22b8fa1cf82",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "additionalData": {
    "projectNames": "test_project2,test_project_123",
    "featureSetIds": "633af8750423a61483680341,633af7b80423a61483680335"
  }
}
feature = fs.features["id"]
feature.description = "test description" # similar will be for other fields update
{
  "seq": "2022-10-04T09:27:33.904475-633af7b80423a61483680335",
  "sourceRequest": "",
  "method": "UpdateFeatureFields",
  "timestamp": "2022-10-04T09:27:33.904475",
  "requestId": "cae1e009-5096-4c47-bf62-8f500ec9a7f1",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "633af7b80423a61483680335",
  "projectName": "test_project_123",
  "featureSetName": "test_fs1",
  "additionalData": {
    "featureName": "id",
    "updatedField": "description",
    "updatedValue": "test_description"
  }
}
fs.add_consumers(["test@h2o.ai"])
{
  "seq": "2022-10-04T09:11:59.740232-633af7b80423a61483680335",
  "sourceRequest": "",
  "method": "AddFeatureSetPermission",
  "timestamp": "2022-10-04T09:11:59.740232",
  "requestId": "b159d492-ca02-4717-814e-16b494963b70",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "633af7b80423a61483680335",
  "projectName": "test",
  "featureSetName": "test_fs1",
  "additionalData": {
    "permissionType": "Consumer",
    "users": "test@h2o.ai"
  }
}
fs.remove_consumers(["test@h2o.ai"])
{
  "seq": "2022-10-04T09:14:07.807106-633af7b80423a61483680335",
  "sourceRequest": "",
  "method": "RemoveFeatureSetPermission",
  "timestamp": "2022-10-04T09:14:07.807106",
  "requestId": "886835ee-bc45-4914-abdc-89317101df14",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "633af7b80423a61483680335",
  "projectName": "test_project_123",
  "featureSetName": "test_fs1",
  "additionalData": {
    "permissionType": "Consumer",
    "users": "test@h2o.ai"
  }
}
fs.ingest(csv)
{
  "seq": "2022-07-20T10:34:17.366153-62d7bd0c93f57739745041a5",
  "sourceRequest": "",
  "method": "StartIngest",
  "timestamp": "2022-07-20T10:34:17.366153",
  "requestId": "301bb8e6-8e64-4d8e-92e6-96a436e29e89",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "62d7bd0c93f57739745041a5",
  "projectName": "test1",
  "featureSetName": "fs_test",
  "additionalData": {
    "jobId": "62d7be0993f57739745041a9"
  }
}
{
  "seq": "2022-07-20T11:45:06.221177-62d7ce6305797005c7dadbad",
  "sourceRequest": "RawData(RawDataLocation(Csv(CSVFileSpec(s3a://feature-store-test-data/smoke_test_data/training.csv,,,UnknownFieldSet(Map()))),UnknownFieldSet(Map())))",
  "method": "EndIngest",
  "timestamp": "2022-07-20T11:45:06.221177",
  "id": "62d7ce6305797005c7dadbad",
  "ingestionCount": 3,
  "ingestionStartTime": "2022-07-20T09:44:40.635830",
  "ingestionEndTime": "2022-07-20T09:44:43.273830",
  "projectName": "test1",
  "featureSetName": "fs_test"
}
{
  "seq": "2022-10-03T16:55:16.215758-633af7c00423a61483680339",
  "method": "JobStatus",
  "timestamp": "2022-10-03T16:55:16.215758",
  "id": "633af7c00423a61483680339",
  "additionalData": {
    "eventId": "00a10969-f48f-4f3e-b125-6071f5c4633f"
  },
  "jobStatus": "Running"
}
fs.get_preview()
{
  "seq": "2022-10-03T17:13:19.862989-633af7b80423a61483680335",
  "method": "GetPreview",
  "timestamp": "2022-10-03T17:13:19.862989",
  "requestId": "94d0f5de-5c34-4197-bce3-8e2480c5ba76",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "633af7b80423a61483680335",
  "projectName": "test",
  "featureSetName": "test_fs"
}
ref.as_spark_frame(spark_session)
{
  "seq": "2022-10-03T17:19:04.772158-633af7b80423a61483680335",
  "sourceRequest": "",
  "method": "RetrieveAsSpark",
  "timestamp": "2022-10-03T17:19:04.772158",
  "requestId": "8ad0f47f-a057-4e29-aeb6-d3e761421fd7",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "633af7b80423a61483680335",
  "projectName": "test",
  "featureSetName": "test_fs"
}
ref.download()
{
  "seq": "2022-10-03T17:14:17.676281-17a91bc8-3733-4d40-af33-e729cce6823a",
  "method": "RetrieveAsLinks",
  "timestamp": "2022-10-03T17:14:17.676281",
  "requestId": "aaf0f2d9-7922-4c00-be28-bf7302d2160d",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "additionalData": {
    "jobId": "633afc1d0423a61483680347"
  }
}
ref = ingest.retrieve()
ref.download()
{
  "seq": "2022-10-04T09:41:23.219196-633be24c708da32636a7cdda",
  "sourceRequest": "",
  "method": "StartRetrieveJob",
  "timestamp": "2022-10-04T09:41:23.219196",
  "requestId": "27568494-7765-494e-9c83-a19ae73c7ddb",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "633be24c708da32636a7cdda",
  "projectName": "test_project",
  "featureSetName": "test_fs",
  "additionalData": {
    "jobId": "633be3a3708da32636a7cde8"
  }
}
ingest.revert()
{
  "seq": "2022-10-04T09:43:51.959670-633be24c708da32636a7cdda",
  "sourceRequest": "",
  "method": "StartRevertIngest",
  "timestamp": "2022-10-04T09:43:51.95967",
  "requestId": "7ec74e9e-660f-40ae-be3a-0a54fc3b9e7a",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)",
  "id": "633be24c708da32636a7cdda",
  "projectName": "test_project",
  "featureSetName": "test_fs",
  "additionalData": {
    "jobId": "633be437708da32636a7cdeb"
  }
}
client.auth.login()
{
  "seq": "2022-10-03T16:42:36.891337-17a91bc8-3733-4d40-af33-e729cce6823a",
  "method": "UserLogin",
  "timestamp": "2022-10-03T16:42:36.891337",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)"
}
client.auth.logout()
{
  "seq": "2022-10-03T16:43:51.724728-17a91bc8-3733-4d40-af33-e729cce6823a",
  "method": "UserLogout",
  "timestamp": "2022-10-03T16:43:51.724728",
  "requestId": "bf4b8a63-1ee1-416e-a347-2c04c7af485b",
  "userId": "17a91bc8-3733-4d40-af33-e729cce6823a",
  "userAgent": "feature-store-py-cli/SUBST_FS_VERSION grpc-python/1.43.0 grpc-c/21.0.0 (osx; chttp2)"
}

Consume events​

All events sent by the core service are in JSON format. The consumers need to pull the data from the topic specified in the configuration.

Kafka has a command line consumer that dumps out messages to standard output.

$ kafka-console-consumer.sh --bootstrap-server localhost:9094 --topic notifications --from-beginning

{
 "seq": String (Surrogate Key: Timestamp-id),
 "sourceRequest": String (in case of a feature set, it contains be data source domains, for ingest it contains ingest source),
 "method": String (name of the action in Feature Store),
 "timestamp": LocalDateTime,
 "requestId": String*,
 "userId": String*,
 "userAgent": String*,
 "id": String* (project or feature set id),
 "ingestionCount": Long* (only available whit ingest statistics),
 "ingestionStartTime": String*,
 "ingestionEndTime": String*,
 "projectName": String*,
 "featureSetName": String*,
 "additionalData": {  }*
}

Keys in additionalData object are:

  • jobId
  • permissionType
  • users
  • featureName
  • updatedField
  • updatedValue
  • message
  • featureSetVersion
  • updatedVersion
  • projectNames
  • featureSetIds
  • eventId
  • description
  • customData
  • accessModifier
  • featureSetType
  • applicationName
  • deprecated
  • dataSourceDomains
  • tags
  • processInterval
  • processIntervalUnit
  • flow
  • featureSetType
  • applicationId
  • approved
  • notes
  • onlineNamespace
  • connectionType
  • topic
  • ttlOffline
  • ttlOfflineInterval
  • ttlOnline
  • ttlOnlineInterval
  • featureId
  • status
  • dataType
  • importance
  • classifiers
  • special

and values are strings.

The method field can have one of the following values:

  • CreateProject
  • DeleteProject
  • ListProjects
  • GetProject
  • AddProjectPermission
  • RemoveProjectPermission
  • OfflineFeatureSetRegister
  • CreateNewFeatureSetVersion
  • DeleteFeatureSet
  • AddFeatureSetPermission
  • RemoveFeatureSetPermission
  • UpdateFeatureFields
  • GetFeatureSetLastMinor
  • GetFeatureSet
  • ListFeatureSetVersions
  • ListFeatureSets
  • UpdateFeatureSetFields
  • StartIngest
  • StartRevertIngest
  • StartRetrieveJob
  • RetrieveAsSpark
  • GetPreview
  • RetrieveAsLinks
  • JobStatus
  • UserLogin
  • UserLogout
  • EndIngest
  • UpdateProjectFields

This value shows which action was triggered by the user.

note

Fields are only available in JSON when not null. For example, the id field, depending on method, will be ProjectId when using Project API.

Feedback