Console app prototype for stream analytics replacement #78

wi-y · 2020-11-21T01:57:15Z

This is the initial prototype for an oss stream analytics replacement console app.

Before running the app, you will also need fill out appsettings.json with the appropriate event hub and storage account information, and well as instruct the app to read from the appropriate event hub (determined by Console:EventHub).

To run the app, navigate to the console folder of the project and set Microsoft.Health.Fhir.Ingest.Console as the startup project, and then build/run. Some logging info will appear in the console for number of events processed for a given window.

You can adjust the window in appsettings.json via EventBatching:FlushTimespan and EventBatching:MaxEvents.

There are some todos remaining which include:

removing hard-coded template files
logging
managed identity (with event hubs, storage accounts, fhir server)
testing min and max batch window interval sizes and queue sizes

src/console/MeasurementCollectionToFhir/Processor.cs

src/console/devicecontent.json

src/console/MeasurementCollectionToFhir/Processor.cs

src/lib/Microsoft.Health.Events/EventConsumers/Service/EventConsumerService.cs

src/lib/Microsoft.Health.Events/EventConsumers/IEventConsumer.cs

src/lib/Microsoft.Health.Events/EventConsumers/Service/EventBatchingService.cs

src/lib/Microsoft.Health.Events/EventConsumers/Service/EventConsumerService.cs

src/lib/Microsoft.Health.Events/EventConsumers/Service/EventBatchingService.cs

src/lib/Microsoft.Health.Events/EventConsumers/Service/Infrastructure/EventQueueWindow.cs

dustinburson · 2020-12-02T04:04:34Z

src/lib/Microsoft.Health.Events/EventConsumers/Service/Infrastructure/EventQueueWindow.cs

+        {
+            while (currentEnqueudTime >= _windowEnd)
+            {
+                _windowEnd = _windowEnd.Add(_flushTimespan);


You should be able to do this with math. The difference between the times will result in a timespan. You should be able to determine how many times bigger the time is from the window end. You can then scale the window end by X amount of the timespan.

src/lib/Microsoft.Health.Events/EventConsumers/Service/Infrastructure/EventQueue.cs

dustinburson · 2020-12-02T04:16:30Z

src/lib/Microsoft.Health.Events/EventConsumers/Service/EventBatchingService.cs

+            {
+                Console.WriteLine($"Threshold wait reached. Flushing {_eventQueues[queueId].GetQueueCount()} events up to: {windowEnd}");
+                var events = await GetQueue(queueId).Flush(windowEnd);
+                await _eventConsumerService.ConsumeEvents(events);


We need to increment the window time here as well correct? If we don't any new event that arrives will immediately trigger a new batch.

I don't think we need to increment here. This method is only called when we haven't received an event in a while, and will only flush if the window end is sufficiently in the past.

If we get a new event then it has to be outside the current window end, we will flush the previous window which has 0 events and advance the window to contain this new event.

The main reason why I didn't want to increment here is to avoid a race condition with the other flushing mechanisms (ThresholdTimeReached and ThresholdCountReached). Perhaps this should be better designed though

Is there a race condition they should be processed on the same thread? We had a race condition in my version since we had a timer triggering the processing.

src/lib/Microsoft.Health.Logger/Telemetry/ITelemetryLogger.cs

dustinburson · 2020-12-14T21:37:43Z

src/console/Normalize/ProcessorStartup.cs

+        public void ConfigureServices(IServiceCollection services)
+        {
+            var outputEventHubConnection = Configuration.GetSection("OutputEventHub").Value;
+            var outputEventHubName = outputEventHubConnection.Substring(outputEventHubConnection.LastIndexOf('=') + 1);


var outputEventHubName = outputEventHubConnection.Substring(outputEventHubConnection.LastIndexOf('=') + 1); [](start = 12, length = 107)

Should be able to use the event hub connection string builder here.

Once we move to Managed Identity, we probably want to have setting for the different components the connection string represents (minus the SAS token):
ConsumerGroup
FullyQualifiedNamespace
EventHubName

Yes, once this is in I am planning on making stories for supporting MI. There are few things I would like to iterate on overall in this code but I would like to this initial build in as a base committed to master so we have the history.

In reply to: 544605846 [](ancestors = 544605846)

dustinburson

@wi-y I re-reviewed. I think you addressed most of the must fix items. I would like to get at least one other person's review here though and get another point of view. It also looks like the build isn't triggering for this PR. Is that something you could look into?

Overall I think we may want to iterate on this some more but rather than do it here I would like to get into master and do follow up with some additional PRs. Make sense?

wi-y · 2020-12-15T18:09:45Z

@wi-y I re-reviewed. I think you addressed most of the must fix items. I would like to get at least one other person's review here though and get another point of view. It also looks like the build isn't triggering for this PR. Is that something you could look into?

Overall I think we may want to iterate on this some more but rather than do it here I would like to get into master and do follow up with some additional PRs. Make sense?

@dustinburson thank you for reviewing. I agree with getting this into master and iterating. I'll look into the builds not triggering.

src/console/MeasurementCollectionToFhir/Processor.cs

src/lib/Microsoft.Health.Events/EventCheckpointing/StorageCheckpointClient.cs

rogordon01 · 2020-12-15T21:03:59Z

src/lib/Microsoft.Health.Events/EventHubProcessor/EventProcessor.cs

+                return Task.CompletedTask;
+            }
+
+            async Task ProcessInitializingHandler(PartitionInitializingEventArgs initArgs)


Simplifying the codebase by using the built in checkpointing sounds like a resonable suggestion. If we find that the extra values being checkpointed (i..e enqueuedDateTime) don't provide value then we could explore a refactor.

src/lib/Microsoft.Health.Events/Repository/StorageManager.cs

namalu · 2020-12-15T00:04:20Z

src/console/MeasurementCollectionToFhir/Processor.cs

+        private ITelemetryLogger _logger;
+
+        public Processor(
+            [Blob("template/%Template:FhirMapping%", FileAccess.Read)] string templateDefinition,


I was required to replace ":" with "_" in order to deploy this in a Linux container.

namalu · 2020-12-16T20:36:39Z

src/console/Normalize/ProcessorStartup.cs

+        public void ConfigureServices(IServiceCollection services)
+        {
+            var outputEventHubConnection = Configuration.GetSection("OutputEventHub").Value;
+            var outputEventHubName = outputEventHubConnection.Substring(outputEventHubConnection.LastIndexOf('=') + 1);


Once we move to Managed Identity, we probably want to have setting for the different components the connection string represents (minus the SAS token):
ConsumerGroup
FullyQualifiedNamespace
EventHubName

src/console/IomtLogging.cs

src/lib/Microsoft.Health.Events/EventCheckpointing/StorageCheckpointClient.cs

namalu · 2020-12-16T23:35:36Z

src/lib/Microsoft.Health.Events/EventCheckpointing/StorageCheckpointClient.cs

+            _publisherTimer.Enabled = true;
+        }
+
+        private async void OnTimedEvent(object source, ElapsedEventArgs e)


I agree with Dustin. I think that the checkpoint should be the completed work "source of truth". Should this process crash for some reason, we could lose checkpoints that were saved in memory. Also, ListCheckpointsAsync, only lists checkpoints that are in blob, and will not include the checkpoints in memory. This seems like trouble.

namalu · 2020-12-17T17:36:34Z

src/lib/Microsoft.Health.Events/Model/EventMessage.cs

+            DateTimeOffset enqueuedTime,
+            IDictionary<string, object> properties,
+            IReadOnlyDictionary<string, object> systemProperties)
+        {


I'm wondering if it makes sense to do null checks here. It seems like some of these properties are "required", so enforcing this on the initializer might make sense. That or maybe in the factory.

src/lib/Microsoft.Health.Fhir.Ingest/Service/MeasurementFhirImportService.cs

dustinburson

* template and script for deploying sa replacement as webjobs * fix template bug * log measurment metrics * fix params * update arm template and deploy script * set alwaysOn * fix errors/telemetry

…github.com/microsoft/iomt-fhir into personal/wiyochum/event-hub-read-and-batch

wi-y added 5 commits August 28, 2020 15:50

method to prefix metric name

3b400d4

Merge branch 'master' of https://github.com/microsoft/iomt-fhir

8449e3a

Merge branch 'master' of https://github.com/microsoft/iomt-fhir

ad66e1f

console app stream analytics replacement

923fff0

remove test app files

b5aab81

wi-y requested review from ms-teli, dustinburson, namalu and rogordon01 November 21, 2020 01:57

dustinburson mentioned this pull request Nov 25, 2020

[Question] What's the purpose of Stream Analytics? #77

Closed

wi-y commented Nov 25, 2020

View reviewed changes

src/console/MeasurementCollectionToFhir/Processor.cs Show resolved Hide resolved

wi-y commented Nov 25, 2020

View reviewed changes

src/console/devicecontent.json Outdated Show resolved Hide resolved

dustinburson requested changes Dec 2, 2020

View reviewed changes

wi-y added 2 commits December 7, 2020 09:42

remove hardcoded mapping files

82824fe

address pr feedback

fa32689

dustinburson reviewed Dec 14, 2020

View reviewed changes

src/lib/Microsoft.Health.Logger/Telemetry/ITelemetryLogger.cs Outdated Show resolved Hide resolved

dustinburson reviewed Dec 14, 2020

View reviewed changes

wi-y added 2 commits December 14, 2020 14:53

refactor and address pr feedback

8181b78

fix template bug

6b016d0

dustinburson self-requested a review December 15, 2020 17:46

dustinburson reviewed Dec 15, 2020

View reviewed changes

rogordon01 reviewed Dec 15, 2020

View reviewed changes

src/console/MeasurementCollectionToFhir/Processor.cs Outdated Show resolved Hide resolved

rogordon01 reviewed Dec 15, 2020

View reviewed changes

src/lib/Microsoft.Health.Events/EventCheckpointing/StorageCheckpointClient.cs Show resolved Hide resolved

rogordon01 reviewed Dec 15, 2020

View reviewed changes

wi-y added 3 commits December 15, 2020 18:41

log measurement metrics

dbbf703

remove timers from checkpointing

521f4ac

testing build pipeline trigger

7cee826

namalu reviewed Dec 17, 2020

View reviewed changes

dustinburson approved these changes Dec 18, 2020

View reviewed changes

wi-y and others added 5 commits January 5, 2021 16:43

Personal/wiyochum/arm deployment event batching (#83)

4d006ae

* template and script for deploying sa replacement as webjobs * fix template bug * log measurment metrics * fix params * update arm template and deploy script * set alwaysOn * fix errors/telemetry

remove debugging

020f629

add null check

ab78a3a

Merge branch 'personal/wiyochum/event-hub-read-and-batch' of https://…

c67bf3d

…github.com/microsoft/iomt-fhir into personal/wiyochum/event-hub-read-and-batch

update namespace

4d3b85f

wi-y merged commit d7fc484 into master Jan 6, 2021

dustinburson deleted the personal/wiyochum/event-hub-read-and-batch branch November 8, 2022 18:34

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Console app prototype for stream analytics replacement #78

Console app prototype for stream analytics replacement #78

wi-y commented Nov 21, 2020

dustinburson Dec 2, 2020

dustinburson Dec 2, 2020

wi-y Dec 3, 2020

dustinburson Dec 9, 2020

dustinburson Dec 14, 2020

namalu Dec 16, 2020

dustinburson Dec 17, 2020

dustinburson left a comment

wi-y commented Dec 15, 2020

rogordon01 Dec 15, 2020

namalu Dec 15, 2020

namalu Dec 16, 2020

namalu Dec 16, 2020

namalu Dec 17, 2020

dustinburson left a comment

Console app prototype for stream analytics replacement #78

Console app prototype for stream analytics replacement #78

Conversation

wi-y commented Nov 21, 2020

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dustinburson left a comment

Choose a reason for hiding this comment

wi-y commented Dec 15, 2020

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

dustinburson left a comment

Choose a reason for hiding this comment