Add realtime inference consumer #90

yatharthranjan · 2021-10-20T14:05:28Z

Realtime Consumers are consumers for the results from the invocation of a model from the realtime analysis pipeline (see model-builder, model-invocation-endpointand ksqldb). The results from the pipeline are in JSON and the invocation is called by a KSQL function which gets inference results then puts the results in a kafka topic. These consumers are meant to consume data from this topic.

3 terms used -

Action: Any task to be performed is an action. This can be sending a notification or an email
or just logging something.
Condition: This is a predicate that will be evaluated on the incoming data. For instance, this
can be used to evaluate if the ML model run crossed a threshold of some sort.
RealtimeInferenceConsumer: This is the base consumer for realtime results topics. Each consumer
can be configured with a set of Conditions and Actions. If all the conditions evaluate to be
true, then all the actions are triggered.

Currently, only one Condition is provided which evaluates a JSONPath expression. This should be sufficient for most simple use cases.
More complex use cases can create concrete implementations of the Condition
interface or ConditionBase abstract class. Currently, the JsonPath expression is read from the
configuration file and hence is static for a particular consumer and topic. Later, we can also
provide this through AppConfig so it can be dynamic based on userId or projectId.

The supported config (for instance an intervention using aRMT app) is of the format -

realtime-consumers:
  - name: 'lstm-lung-study-consumer' # Name of the consumer
    topic: 'lung_study_lstm_ad_inference' # Kafka topic to consume from (this should contain the inference results)
    notify_errors:
      email_addresses:
        - '[email protected]'
    conditions:
      - name: 'LocalJsonPathCondition' # Name of the condition
        projects: [ 'radar-test' ]  # Only evaluate for these projects
        subjects: [ 'sub-1', 'sub-2' ]  # Only evaluate for these subjects
        properties:
          jsonpath: '$[?(@.invocation_result.anomaly_detected == true)]' # JsonPath expression to evaluate
          key: 'invocation_result' # Key that contains data to evaluate 
    actions:
      - name: 'ActiveAppNotificationAction' # Name of the action
        projects: [ 'radar-test' ] # Only execute for these projects
        subjects: [ 'sub-1', 'sub-2' ]  # Only execute for these subjects
        properties:
          questionnaire_name: 'ers' # Name of the questionnaire to trigger
          time_of_day: '09:00:00' # Local user time of day to trigger at
          default_timezone: 'Europe/London' # Default timezone to use for the time of day if not found in the appserver
          appserver_base_url: 'http://localhost:8080/' # Base URL of the appserver
          management_portal_token_url: 'http://localhost/managementportal/api/oauth/token' # URL to get the management portal token
          client_id: 'realtime_consumer' # Client ID for the management portal
          client_secret: 'secret' # Client secret for the management portal
          metadata_key: 'invocation_result' # Key that contains the metdata to be forwarded to the aRMT app
      - name: 'EmailUserAction'
        projects: [ 'radar-test' ]
        subjects: [ 'sub-1', 'sub-2' ]
        properties:
          email_addresses: [ '[email protected]' ]

Note: The properties section is specific to each Action and Condition. Please take a look at the
condition and action docs for the keys supported. If the projects or subjects key is not
specified the action and condition will be used on all projects or subjects respectively.

Release 0.3.0

Release 0.4.0

- Includes various implementations includeing appserver notification actions and email user. - Conditions evaluation using json path (currently locally, later read jsonpath from appconfig too).

blootsvoets

Yes this looks like a very good start! At some point the threshold may need to come from radar-app-config service instead of from a fixed value. This may also be a good entry point to enable / disable the condition for a given project (e.g. to only process records for certain projects).

blootsvoets · 2021-10-21T07:28:08Z

src/main/java/org/radarcns/consumer/realtime/action/ActiveAppNotificationAction.java

+                .getProperties()
+                .getOrDefault(
+                    "management_portal_token_url",
+                    "https://radar-cns-platform.rosalind.kcl.ac.uk/managementportal/api/ouath/token");


Preferably keep this out of the code. If you want to provide defaults, it would be better to define a base URL somewhere, and use that plus the default path as the default.

blootsvoets · 2021-10-21T07:28:56Z

src/main/java/org/radarcns/consumer/realtime/action/EmailUserAction.java

+        new EmailSender(
+            (String) props.getOrDefault("host", "localhost"),
+            (Integer) props.getOrDefault("port", 25),
+            (String) props.getOrDefault("from", "[email protected]"),


This should not have a default.

blootsvoets · 2021-10-21T07:30:20Z

src/main/java/org/radarcns/consumer/realtime/action/appserver/AppserverClient.java

+    objectMapper = new ObjectMapper();
+
+    if (tokenUrl == null || tokenUrl.isEmpty()) {
+      logger.warn("MP Token url was not provided. Will use unauthenticated requests to appserver.");


Is this ever the case? It would be better to exclude this.

Unlikely in a production environment. The appserver does support a no-security mode where unauthenticated requests can be made (particularly useful for dev environments). I will remove it once I have tested it with the appserver.

yatharthranjan · 2021-11-30T16:13:28Z

At some point the threshold may need to come from radar-app-config service instead of from a fixed value.

Yes definitely can create impl to read the config from the app-config.

This may also be a good entry point to enable / disable the condition for a given project (e.g. to only process records for certain projects).

Yes a good idea, will have a think through how it can be done, since then conditions will need to be accessed based on the project in the record in real-time. I guess a very simple first approach could be to just to add a getConditionsFor(String project) and getActionsFor(String project) and possibly also a user-specific variant like getConditionsFor(String project, String user) to the RealtimeConsumerConfig, and use that on each incoming record.

EDIT: Actually made projects a part of conditions and actions, so you can specify a projects list to filter-

...
     conditions:
        - name: 'LocalJsonPathCondition'
          projects: ['radar-test']
...

If no projects are provided, works as default on all projects. But now I am wondering how this will be achieved when we move to app-config - will the conditions have the projects key? Or will it be on the request path like /config/client/realtime-consumers/projects/{project} ?

nivemaham · 2022-01-12T20:51:45Z

src/main/java/org/radarcns/consumer/realtime/RealtimeInferenceConsumer.java

+ * <p>To be used with the model-invocation-endpoint and KSQL API_INFERENCE function to evaluate and
+ * take action on incoming results from realtime inference on data.
+ */
+public class RealtimeInferenceConsumer implements KafkaMonitor {


If I understand it correctly, all of these Consumers are realtime. We can remove that word from the class name I think.
Why not reuse/extend from existing AbstractKafkaMonitor?

I think the adjective realtime was added for inference and not for consumers. So these are consumers for realtime inference as compared to batch or historic inference. But I agree if it seems confusing best to remove it. AbstractKafkaMonitor was not extended just to allow more flexibility (like for now a different constructor spec) but can definitely extend from AbstractKafkaMonitor if we need functionality already present there.

nivemaham · 2022-01-12T20:53:11Z

src/main/java/org/radarcns/consumer/realtime/RealtimeInferenceConsumerFactory.java

+            .map(c -> createConsumer(handler.getRadarProperties(), c)));
+  }
+
+  private static KafkaMonitor createConsumer(


I would call it InferenceMonitor, InferenceMonitorFactory for consistency

nivemaham · 2022-01-12T20:57:34Z

src/main/java/org/radarcns/consumer/realtime/action/appserver/AppserverClient.java

+    }
+  }
+
+  public Map<String, Object> createMessage(


you are referring to the create operation that is done on the service that receives the request. Right? I think this should be either sendMessage or addMessage.

yes good point

Also replaced all deprecated code

- Add realtime consumer to docker builds for integration tests - Fix starting for docker application when no schemas in schema registry - Updates classes and code based on expected API spec of appserver - Reuse classes wherever possible - Fix timezone conversions and add jitter and tolerance to the time calculations - Simplify configs and string formatting - Handle unhandled exceptions

- Add more properties and fixes - update fields and time calculations based on app expectations

- Add smtp to docker-compose.yml - Also catch all exceptions from action execution, so other actions can proceed.

Finalise realtime inference

…usly - Make delay configurable

- This is to prevent duplicate notifications

yatharthranjan · 2024-09-24T15:42:58Z

@mpgxvii, is it possible for you to review this when you have some free bandwidth? I think you plan to use this in the near future, so might be good to review and familiarise, unless you plan to use something else?

yatharthranjan and others added 7 commits July 17, 2018 10:44

Merge pull request #77 from RADAR-base/release-0.3.0

b5191c6

Release 0.3.0

Merge pull request #83 from RADAR-base/release-0.4.0

36e8ae2

Release 0.4.0

Initial impl for realtime consumers

d549524

Fix dependencies

850c0e6

First complete set of conditions and actions

3109e31

- Includes various implementations includeing appserver notification actions and email user. - Conditions evaluation using json path (currently locally, later read jsonpath from appconfig too).

better use of enums

70f1594

add basic javadocs

1d5915a

yatharthranjan requested review from blootsvoets and nivemaham as code owners October 20, 2021 14:05

yatharthranjan marked this pull request as draft October 20, 2021 14:05

simplify defaults and enum

be852c8

blootsvoets reviewed Oct 21, 2021

View reviewed changes

yatharthranjan and others added 3 commits November 30, 2021 17:48

add support for filtering conditions and actions based on projects.

bc8cbc8

remove unintended check

9b08b8a

add appserver related components to docker-compose.yml file

57c246f

nivemaham reviewed Jan 12, 2022

View reviewed changes

yatharthranjan and others added 11 commits January 26, 2022 12:19

Merge remote-tracking branch 'origin/master' into add_realtime_consumer

72ac511

minor fixes to make build successful

03f3762

Merge branch 'add_realtime_consumer' into intervention-trigger

9fb26af

resolve conflicts

2b53ffc

move monitor configs to separate package

50af14f

rename NotifyConfig.java as EmailNotifyConfig

a93afd8

restructure email server config.

e1a3b93

new intervention config model

25435cb

Commit InterventionMonitor

426f4e6

Commit config format and object creation

077837b

Implemented intervention monitor and updated build config

fac723b

Also replaced all deprecated code

blootsvoets and others added 18 commits February 15, 2022 15:35

Fix docker setup

628d17a

Use sh in docker

6522391

Move email senders and simplify config

5811215

Tested and fixed tests

3e199ef

Tested full flow

9c25663

Rename .java to .kt

618c9e7

Initial kotlin convert of added classes

06e5482

various fixes and improvements

31af146

Minor updates

e7a8b9b

Add localisation support for AppServerIntervention.kt

4e92616

- Add more properties and fixes - update fields and time calculations based on app expectations

fix referenceTimestamp

a24adcf

exclude clinicalProtocol if null

ed04048

fix eval

77a8082

fix and test email action

0a745ce

- Add smtp to docker-compose.yml - Also catch all exceptions from action execution, so other actions can proceed.

Notify admin in case of action, condition or consumer failure

17308f3

Add docs on realtime inference consumers

3d808c6

Merge pull request #92 from RADAR-base/finalise_realtime_inference

bfa015c

Finalise realtime inference

yatharthranjan changed the title ~~WIP: Add realtime consumer~~ Add realtime consumer Feb 24, 2022

yatharthranjan marked this pull request as ready for review February 24, 2022 12:37

yatharthranjan changed the title ~~Add realtime consumer~~ Add realtime inference consumer Feb 24, 2022

yatharthranjan added 7 commits February 28, 2022 18:03

JsonPathCondition.kt: return false if record value is null

1fc45c6

Add support for sending both notification and data message simultaneo…

3c5b488

…usly - Make delay configurable

send data messages with HIGH priority

76ab324

make jitter configurable and null by default

28a5874

- This is to prevent duplicate notifications

Fix ttl

0b306f3

add explicit "type" to protocol

497f1ed

Add action for adding triggered task to user's schedule on appserver

41f3582

Merge branch 'dev' into add_realtime_consumer

dde1b4a

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add realtime inference consumer #90

Add realtime inference consumer #90

yatharthranjan commented Oct 20, 2021 •

edited

Loading

blootsvoets left a comment

blootsvoets Oct 21, 2021

blootsvoets Oct 21, 2021

blootsvoets Oct 21, 2021

yatharthranjan Nov 30, 2021

yatharthranjan commented Nov 30, 2021 •

edited

Loading

nivemaham Jan 12, 2022

yatharthranjan Jan 26, 2022 •

edited

Loading

nivemaham Jan 12, 2022

nivemaham Jan 12, 2022

yatharthranjan Jan 26, 2022

yatharthranjan commented Sep 24, 2024

Add realtime inference consumer #90

Are you sure you want to change the base?

Add realtime inference consumer #90

Conversation

yatharthranjan commented Oct 20, 2021 • edited Loading

blootsvoets left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yatharthranjan commented Nov 30, 2021 • edited Loading

Choose a reason for hiding this comment

yatharthranjan Jan 26, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yatharthranjan commented Sep 24, 2024

yatharthranjan commented Oct 20, 2021 •

edited

Loading

yatharthranjan commented Nov 30, 2021 •

edited

Loading

yatharthranjan Jan 26, 2022 •

edited

Loading