SaFi Bank Space : One-Stop Platform Logging for FeatureStore

Overview

The FeatureStores in this page are derived from AAI’s OSP Logging Functionality.

We are using the Logging functionality to receive the output response at each node from AAI, in order to ingest into our DataLake and for our FeatureStore

Whenever a workflow is created, we will need to create Logging Callback URL to receive the logs, and AAI will help us to configure this Callback to the said workflow. For now, only one Callback URL can be configured per workflow. By Nov 2022, AAI said that each individual node in a workflow can have its own Callback URL.

Ideally, there will be dedicated functions to process each type of node. For example, Email Detection API Service will have a specific function to process all logs from Email Detection nodes across different workflows.

Authentication and Tyk URL Mapping

See: Tyk API Gateway for Cloud Functions

Proposed Setup

This is the proposed setup when we can log each nodes individually:

Ingestion Logic from Logging to FeatureStore

The following steps are the workflow as defined in the Entry Point function for the Cloud Function. The current entry point for Cloud Functions used in Logging ingestion is write_logging_request_api()

To see a documented sample, see the following Code for ingestion identity check logs from OSP in the following Github Repo - DataTest/services/cloud_functions_stream/src/main/stream-logging-identitycheck-fswrite/main.py

	Steps	Description	Functions/Methods
1.	Parse from Logging Request	To parse the incoming logging request using `request.get_json()`	`request.get_json()`
2.	Extract relevant features into a dictionary from Request JSON	Transform Request into a dictionary of `{feature_id: feature_vaule}` into key-value pairs for Online FeatureStore ingestion	`ingest_stream_XXXXX()` XXXXX indicates the function name, for instance `ingest_stream_identity_check`
3.	Cloud Authentication	To retrieve the Google Cloud Bearer Token for authentication, in making POST API requests to the endpoint for Online FeatureStore ingestion
4.	Write to Online FeatureStore (REST API)	Create the necessary headers for the POST API request	credentials, project = google.auth.default() auth_req = google.auth.transport.requests.Request() credentials.refresh(auth_req) id_token = credentials.token headers = { "Authorization":"Bearer {}".format(id_token), "Content-Type":"application/json" }
		Create the `featureValues` by passing the above dictionary into a JSON with proper schema to be used as the “data” body of the REST API payload	featureValues = get_featureValues( data, atomic_features = feature_list, output_format="REST", entity_id_name=ENTITY_ID_NAME )
		Post the response to the endpoint	response = requests.post( endpoint, data = json.dumps(featureValues), headers = headers )
5.	Print Status Code of Ingestion Job	Returns a 200 if the ingestion job to online featurestore is successful. Otherwise, [400, 403, 500] are the other typical responses	regex_search = re.compile('entityTypes/(.+?):writeFeatureValues') response_ls.append((response.status_code, re.search(regex_search, endpoint).group(1))) if response.status_code == 200: print("SUCCESS - {} - Written to {}: {}".format(data[ENTITY_ID_NAME], FEATURESTORE_ID, entity_type_id))

Reference Documents

Log Event Sample

{
  "transId": "6be712d68c64058b",
  "flowId": "104",
  "node": {
    "id": 28907,
    "name": "8:Blacklist Check",
    "startTime": "2022-08-12T10:12:16.827Z",
    "endTime": "2022-08-12T10:12:16.828Z",
    "opTime": "2022-08-12T10:12:16.828Z",
    "cacheHit": true,
    "input": [
      {
        "name": "name",
        "value": "SITI CHURIAH"
      },
      {
        "name": "idNumber",
        "value": "3173054609790011"
      },
      {
        "name": "phoneNumber.countryCode",
        "value": "+62"
      },
      {
        "name": "phoneNumber.areaCode",
        "value": ""
      },
      {
        "name": "phoneNumber.number",
        "value": "85775095213"
      }
    ],
    "output": {
      "code": "SUCCESS",
      "message": "OK",
      "data": {
        "recommendation": "REJECT",
        "defaultListResult": [
          {
            "eventTime": "02/2019",
            "productType": "INSTALLMENT_LOAN",
            "hitReason": "IDNumberAndMobileAndName",
            "reasonCode": "OVERDUE_DAYS_GREATER_THAN_90"
          },
          {
            "eventTime": "10/2018",
            "productType": "CASH_LOAN",
            "hitReason": "MobileAndName",
            "reasonCode": "OVERDUE_DAYS_BETWEEN_60_TO_90"
          },
          {
            "eventTime": "10/2018",
            "productType": "CASH_LOAN",
            "hitReason": "MobileAndName",
            "reasonCode": "OVERDUE_DAYS_BETWEEN_30_TO_60"
          }
        ]
      },
      "extra": null,
      "transactionId": "84e32dcfecc9f4dc",
      "pricingStrategy": "PAY"
    }
  }
}

SaFi Bank Space : One-Stop Platform Logging for FeatureStore

Attachments: