Orleans: Event Hub Stream Provider instance configuration

[centur]

Hi, seems that there is no documentation in any form for configuring EventHub Stream Provider instances :(. I'd like to use this issue as a starting point and going to throw\document findings I'm scraping from orleans unit tests approach. Also I have some questions already about way forward w.r. to strongly typed vs weakly configurations for Stream Providers.

_Any help, hints and kicks are appreciated ;)_

Question1: Configuration approach shifted from Dictionary<string, string> to EventHubStreamProviderSettings , Lazy<EventHubSettings> and EventHubCheckpointerSettings with their methods WriteProperties(settings) which are writing key value pairs into passed dictionary. This is great bit and makes configuration a bit more friendly, but the question is - can we rely on the fact that the contract of these classes won't be changed and this approach would stay in Orleans ? If not sure - I'd rather prefer to make my own copies of these classes but this implies I have to copy Dictionary keys too... Is this a new way forward with configuration bits ?

[centur]

Question2: for a single stream provider type we need 3 different configuration objects (mentioned above). Is there a reason why they made as a separate classes rather than just a single, bigger but easier to maintain? Also there is a take to resolve these configuration from Dependency injection container by type name and type is specified in the property. Why not to use an interface for that ?

[centur]

Question3 (from gitter): What's the difference between values of this enum
C# public enum StreamQueueBalancerType { ConsistentRingBalancer, DynamicAzureDeploymentBalancer, StaticAzureDeploymentBalancer, DynamicClusterConfigDeploymentBalancer, StaticClusterConfigDeploymentBalancer, }
I've read the comments on the members but they don't make much sense for me.
Can anyone explain in simple terms when I should prefer ConsistentRing and when Dynamic Azure\Cluster Configuration ? What is ups and downs of each type ? It seems that when running in Azure Cloud Service ( not SF) it's reasonable to set to DynamicAzureDeploymentBalancer , how it affects Event Hub behaviour ?

[gabikliot]

Re last question: how many eh queues do you have and how many silos? The answer will impact what balancer type to use.

[centur]

@gabikliot I'm not sure what queues means in this context - Event Hub has Partitions. Are they implied by Queues ? Silo number is dynamic - we can scale out and in when needed, so it's not a fixed number. We currently have 2 and as suggested we picked large vm instances. I'm also not sure about balancing - from what I groked about EH - number of partitions should be the upper limit of concurrent readers in a single consumer group (e.g. one reader per partition). Does this means that Orleans can guarantee and control this behaviour ?

[centur]

Question4: there is a different SAS keys can be created to access EH - key for writing and key for reading. Connection string that EHSP requires - does it implied to have Send and Listen claim on the same key? What's about Client configuration - do I need to configure the OrleansClient with the same key or I can limit it only to Send permission ?

[gabikliot]

Yes, I saw asking about typical number of partitions and typical to your deployment number of silos.

[centur]

@gabikliot a bit of a side question - correct me if I'm wrong - the scaling of EH happens on EH Namespace level - one can increase number of throughtput units there. Does this means that all Hubs inside the namespace will share and collaborate over the Namespace throughput resources ? How does it work with EHSP then ? Suddenly there may be a drop in throughput if some other Hub in the same namespace would get a sudden spike, right ?

[gabikliot]

I don't know. I can tell you about the balancer, if you want to know that.

[centur]

@gabikliot any info would be handy as it will make your answer searchable on the github =)
I understand the ConsistentRing - this will probably balance queue (readers?) across all siloes using that ring hashing approach, right ?

[gabikliot]

Can u please tell the expected number of partitions and silos?

[centur]

dev-test env - 2 partitions and 3 small silos
Prod probably starts with 8 partitions (as it can't be changed lately) , starts with 2 silos, may scale out dynamically to 8. We never saw it scale higher than 3 though (and most often - webrole scales out, not a silo worker), but we haven't had high load yet.

[jason-bragg]

As @gabikliot seems to be addressing the queue balancing questions, I'll start with other questions. Though I can chime in on that subject if needed.

Configuration

Can we rely on the fact that the contract of these classes won't be changed and this approach would stay in Orleans.

For the most part yes, but not in any absolute sense. The EHStreamProvider is new and streaming itself is not a solved problem. While there are a number of production services using this tech, and we've no planned changes, I don't feel confident claiming that it is written in stone. With the exception of the EventHubStreamProviderSettings, most of the related configuration components implement interfaces to support integration with your services existing configuration system, so the system is very flexible, but my recommendation would be to stick with the existing classes until specialization is needed.

Is this a new way forward with configuration bits ?

TL;DR : No. not at this time.
Long version - I chose to error on the side of modularity, customizability, and extensibility, with the intention to build more user friendly façades over this complexity. I've, unfortunately, not yet added the user friendly façades. I was given the freedom to make these decisions and hoped that the final result (which has not yet been achieved) would serve as a demonstrably superior approach to the previous configuration patterns. If successful, these patterns will influence future configuration patterns used in Orleans, but only if successful.

Is there a reason why they made as a separate classes rather than just a single, bigger but easier to maintain?

Modularity. Each configuration interface relates to a distinct and coherent set of settings. My intention (not yet achieved) was to allow for independently configurable components, and have a convenience class that implements all of the interfaces.

resolve these configuration from Dependency injection container by type name and type is specified in the property. Why not to use an interface for that ?

An interface would limit these to one instance per system, which only works in services with a single eventhub stream provider. Using the resolver was a request by a consumer that wanted to use their own configuration system along with DI to inject the settings, but I'm not pleased with the final outcome, in part because our DI system is not terribly mature.

[jason-bragg]

SAS keys and read write permissions

does it implied to have Send and Listen claim on the same key?

On the silo, both read/write permission are required.
Client needs only write permission.

[jason-bragg]

help, hints and kicks

The EventHubStreamProvider is the most customizable stream provider we support. This was done to maintain support for 343 industries features (very important as they were kind enough to contribute the provider). This comes at the cost of complexity, which I hope to eventually hide under a user friendly façade, but as that has not yet been done, learning curve may be non-trivial.

The majority of the production services use the EventHubStreamProvider to read and process events from eventhub, few use it to write events and those do not do so at rates comparable to the supported read rate (2k events/sec/silo). This means that the read pipe has been hardened and tuned far more than the write pipe. If your service requires a write rate comparable to the supported read rate, understand that the tech has not been demonstrated to support those numbers, and that developing (and ideally contributing) optimizations in this area will likely be necessary.

Orleans serialization does not support backwards compatibility, so EHSP, out of the box, does not either. This means that events written to eventhub cannot be changed. New events can be added, but if the data type of events written to eventhub change, they will no longer deserialize. If backwards compatibility is required, developers will need provide their own data adapter that serializes data in a version tolerant way (bond, protobuf, ..?)

[centur]

on Orleans serialization and EH compatibility - I understand that, we used to this fact already (this why we are heavily "JSONed", despite it may be sub-optimal in some areas). If I understand correctly - the lifetime of an event is up to 7 days (we configured to 1) in EH so old formats will die eventually (although data will be lost). I'm still in research mode about EventHub and the way we want to use it, but for now we will have a http frontend that will do all the writes so API consumers can stay on the HTTP interface they used to and then eventually transition to direct EH writes for certain areas, if it's feasible for their platform.

So far our write rate will be equal to our read rate (the single event will be read once in EH and then distributed to all listeners, same as in azure queues) but we can do direct writes too if needed, so I don't think it'll be a bottleneck at all.

Thanks for the hint about potential option to have multiple EH streams in one app - I'm thinking about a possibility of having "blue-green" hubs in one app and then migration of the data formats will be easier - new formats can be written to a new hub until old formats in old hub won't be fully read and then old hub will stay empty while new will do the job, then in case of another format change - we would swap them again.

[jason-bragg]

help, hints and kicks

One more.. the EHSP does not create the consumer group, so you'll need to create that either when you setup the hubs or during some service startup logic (prior to silo initialization). This is by design, not an oversight.

Multiple EH per service
There are many reasons to have multiple providers and it is fully supported, however, queue balancing has not yet been optimized to handle multiple providers as well as we'd like. Using any of the queue balancers other than ConsistentRingBalancer, the balancing algorithm is consistent across all stream providers, so small imbalances can be amplified with multiple providers configured. Consider a hub with 4 partitions in a 3 silo cluster. Distribution may be 2, 1, 1, but if 10 similar stream providers are configured the distribution will be 20, 10, 10, rather than the desired 14, 13, 13. Think of these little imperfections as opportunities to give back to the community :)
_Edited to fix queue balancing math. Apparently I can't count._

[centur]

Ok, this is what I discovered so far:
EventHubStreamProvider consists 2 main moving parts to be configured -

• EventHub itself - details on how to connect to EventHub instance. Configured with EventHubSettings class
• Event Hub Checkpointer - the mechanism that allows to continue reading from a checkpoint in time from EventHub Partition (rewindable stream), configured with EventHubCheckpointerSettings class

And we still need to add typical StreamProvider configuration itself (using EventHubStreamProviderSettings class)

EventHubSettings ctor require the following parameters
connectionString - your Event Hub Namespace connection string, not a Hub string.
consumerGroup - Event hub concept of consumer groups - when EH is created - there is one consumer group created automatically with the name $Default - can be used out of the box
path - this is tricky one - this is your Event Hub name that created in the Namespace (also this why one need to supply Namespace connection string, not a Hub connection string)

Non-mandatory params are bit more self explaining:
startFromNow is a flag that indicates that reading must happen from current time (and not to peek in the past data). I'm not sure how it suppose to work with checkpointer, it seems that it affects only initial behaviour and if there is a checkpointer record - it will be used instead of reading from "now"...
prefetchCount - number of prefetched events on read ( set to EventHubConsumerGroup.PrefetchCount, see this )

EventHubCheckpointerSettings ctor require the following parameters
dataConnectionString - connection string to Azure Table Storage where the checkpoint table will be created
table - name of the table to create
checkpointNamespace - a suffix that will be used in PartitionKey of the record e.g. partition key is EventHubCheckpoints_BullclipEventHub_checkpoint, namespace was set to BullclipEventHub (rowkey is partition_0 and row itself stores an Offset value of the consumer group)

Non-mandatory params:
persistInterval - how frequent checkpoint data will be saved

So if anyone wants to config , he'd need to do the following:

Cluster configuration would look something like this:
```C#
// StreamProvider settings - type of pubsub and queue balancer type
var settings = new Dictionary {
{ PersistentStreamProviderConfig.STREAM_PUBSUB_TYPE, nameof(StreamPubSubType.ImplicitOnly) },
{ PersistentStreamProviderConfig.QUEUE_BALANCER_TYPE, nameof(StreamQueueBalancerType.ConsistentRingBalancer) }
};

// Ctor with just a name of provider
new EventHubStreamProviderSettings(name)
.WriteProperties(settings);

// Event Hub configuration
new EventHubSettings(eventHubConnectionString, eventHubDefaultConsumerGroup, eventHubName)
.WriteProperties(settings);

// Checkpointer configuration
new EventHubCheckpointerSettings(dataConnectionString, eventHubCheckpointsTableName, ehCheckpointTableNamespace, TimeSpan.FromSeconds(3))
.WriteProperties(settings);

config.Globals.RegisterStreamProvider(name, settings);
```

[centur]

IMO: A bit dubious decision to use strongly typed configuration objects with get only properties.
I'd rather prefer to have validation in the WriteProperties method or in a separate Validate(), than mandatory parameters in ctor and validation on initialization.
Validation on construction is a good pattern to a certain point, but if one would need to have 10 parameters for another stream provider - ctor would be ugly and very unreadable. Also I can
Also for some cases it's handy to serialize these pre-made configuration objects and simplest approach is to have parameter-less ctor and non-private set on properties.

Also one class (EventHubStreamProviderSettings)is made in one way - with defaults and setters and it feels much more readable (IMO):

new EventHubStreamProviderSettings(name)
{
  CacheSizeMb = 128,
  DataMinTimeInCache = TimeSpan.FromSeconds(15)
}
.WriteProperties(settings);

Overall, I like this approach and the DI attempt ( which I'd make differently with a named resolution here, rather than a type driven, but this is a personal preference)

[galvesribeiro]

What if we make all this discussion a doc page? Are you willing to do that @centur?

[jason-bragg]

which I'd make differently with a named resolution here

Our DI is limited to the abstractions provided by asp.net dependency injection. I didn't see any obvious way of doing this. I may well be unaware, so if you can help me understand how to do named resolution under those constraints, I'd prefer that as well.

get only properties

In general, and admittedly debatable, when dealing with configuration data, I prefer immutable data unless the underlying system supports dynamic changes. Some systems need support real time modification of a configuration setting, say logging level for instance. In these cases I allow setters as well. Allowing setters on configuration that is not modifiable makes supported usage patterns less ambiguous. This being said, I will consider your suggestion, as the intended usage patterns should be communicated by the interface, not the implementation, so it's not unreasonable (it's arguably common) to have a mutable configuration object implementation,

As I attempted to articulate earlier, and may have failed, a user friendly façade that makes these component configurations more usable has not yet been introduced. At this layer usability is not the primary requirement. Your feedback is encouraged and will be considered, but where it may have more influence is in suggestions of how to structure the façade. Two paths I've contemplated, but not settled on are:

• Fluent configuration builder - https://msdn.microsoft.com/en-us/library/ff664363(v=pandp.50).aspx
• Extension function like AddAzureTableStorageProvider.

Also open to other suggestions.

[galvesribeiro]

I think configuration at the higher level will be revamped once we start working on SiloBuilder. I've seen multiple cases recently where this story should come first, just like this one.

[centur]

@galvesribeiro - I can, don't want to step on @jason-bragg toes if there is a work in progress for this already

@jason-bragg yeah, just checked the capabilities of DI-abstraction and really frustrated with them. Out of all jack-of-all-trades DI Containers, DI.Abstractions followed the typical known anti-pattern path and became universal master of nothing. I thought that there is a way to do a named resolution like in Autofac - e.g. I'd use name of the stream provider as a configuration retrieval key so in the configuration class you can do something like container.ResolveNamed<IEventHubSettings>("ehStreamProviderName"), so we can register unique settings per provider name, another example is in Autofac doco. Don't take it as a critique, I wasn't aware about such limited capabilities of core DI abstractions. Maybe in the future Core DI abstraction will add extra resolution methods to support such scenario, but current state is a route to build crutches.

[rant]
Sometimes I feel that being opinionated and taking a dependency on say specific serialization library (Json.NET), logger (Serilog) or DI container (Autofac) would provide much more value and allow code to be very elegant and simple. All attempts to build an all-covering abstractions that I saw either failed or heavily leaking...
[/rant]

Regarding immutability of configuration objects - as I said - it's my personal opinion, so take it with a grain of salt. I'm not a big fan of sticking some strict approaches (immutability of config DTO) in the areas that are not implies dangers of changing on the fly, incapsulation would give a certain guarantees that there will be not much code that will need an access to the same object (and it's more a violation of SRP to reuse config object for different domains), also most of the services I saw usually configured once per it's lifetime and then not changing the configuration. It's easier to create a new service with new configuration than try to carefully and reliable reconfigure existing one with new connection strings when it's actively being used by other code.

I leaning towards the configuration extension functions. Before it was added to the Orleans, we wrote set of extensions with our own defaults for the project and I feel that configuration looks pretty flexible and neat:
```C#

config
            .MemoryStorageProvider("MemoryStore")
            // Switch PubSubStore to in-memory provider- as we have cluster specific clients (websockets) there is no point in keeping the subscribers state in the persistent store 
            //See https://github.com/dotnet/orleans/issues/2418 
            .MemoryStorageProvider("PubSubStore") 
            .TableStorageProvider("AzureTable", data.DataConnection)
            .TableStorageProvider("DataStorage", data.DataConnection)
            .ExceptionlessBlobStorageProvider("BlobStorage", data.DataConnection)

            //.ExceptionlessBlobStorageProvider("PubSubStore", data.DataConnection)
            //.ExceptionlessBlobStorageProvider("AzureTable", data.DataConnection)
            //.ExceptionlessBlobStorageProvider("DataStorage", data.DataConnection)
            //.MemoryStorageProvider("BlobStorage")

            .SimpleMessageStreamProvider(Key.SimpleMessagingProviderName)
            .AzureQueueStreamProvider(Key.AzureQueueStreamProviderName, data.DataConnection, queuePrefix: $"streams-{data.DeploymentId}")
            .AzureQueueStreamProvider(Key.WebSocketStreamProviderName, data.DataConnection, "ExplicitGrainBasedAndImplicit", $"sockets-{data.DeploymentId}")
            .SimpleMessageStreamProvider(Key.WebSocketSimpleStreamProviderName, "ExplicitGrainBasedAndImplicit")

            .EventHubStreamProvider(Key.EventHubStreamProviderName,data.DataConnection,
            eventHubConnectionString: "Endpoint=sb://snip.servicebus.windows.net/;--SNIP--", 
            eventHubName:"snip-hub"
            );

```

Fluent builder is an option too, for small sets of parameters. But IMO it's a mess of ifs when you need to create the configuration object based on many string values from a config file or environment vars...

[jason-bragg]

work in progress

None and to be honest, I'm a -horrible-writer, so I'll always be happy to support community adding to the docks (he says almost giddy about the idea).

rant

Don't get me started!
[rant]
_deleted_
[/rant]

configuration extension functions

That's our current pattern, and is not bad. Stream providers are exceptionally complex, (EHSP doubly so) hence the consideration of other approaches.

Thanks for feedback and suggestions. Area needs some attention and outside voices help remind me how much work remains.