MassTransit for Service Fabric

[kevbite]

It would be good if we could get MassTransit working native with Service Fabric, I recently did a little spike to just get a basic ServiceListener working to Start/Stop the Bus within MassTransit which gets a small step in the door, This spike can be found here with an example.

It however might be worth considering:

Partitioning

Within stateful services you can partition your requests based on a partition key, it would be good to get MassTransit hooked in to this so it works seamlessly together

State

Within stateful services we can store isolated state which would be ideal for MassTransit persistence, It's just a dictionary and supports transactions.

[kevbite]

I could try getting the state side of things working but I'm not too sure how we'd tackle partitioning.

[maldworth]

@kevbite , thought I'd update this with my early findings.

First, the MassTransitListener you created in the example is exactly what I used. However you are using a Stateful Service, whereas I used a Stateless. I guess it really depends on what your intentions are here, but either work.

Topshelf.ServiceFabric?

Since my service was made with Topshelf (and Topshelf.Azure), I tried to look at the possibility of Topshelf.ServiceFabric.

Every time I looked at the lifecycle events of Service Fabric (being asynchronous), it didn't feel right tying in the topshelf synchronous start/stop with the Service Fabric asynchronous lifecycle methods (openAsync, closeAsync and abort).

Trying to use Topshelf would be...
__openAsync -> topshelf.run/start -> which is calling bus.start__ (and bus.start is really an extension method with the utils.awaiter for the asynchronous bus startAsync method). This didn't feel quite right.

Another problem is TopShelf needs to run with:

return (int)HostFactory.Run(Configure);

but service fabric needs a _self hosted_ executable.

ServiceRuntime.RegisterServiceAsync("MyServiceType", context => new MyService(context)).GetAwaiter().GetResult();

ServiceEventSource.Current.ServiceTypeRegistered(Process.GetCurrentProcess().Id, typeof(MyService).Name);

// Prevents this host process from terminating so services keep running.
Thread.Sleep(Timeout.Infinite);

It is much easier to call the bus startAsync from the MassTransitListener (as you have done).
So to start Topshelf locally, I just decided to use #ifdef. The
code is nearly the same between MyTopshelfService and MyService below, where one is synchronous and
the other is asynchronous (respectively).

internal static class Program
{
    static void Configure(HostConfigurator hostConfigurator)
    {
        hostConfigurator.Service<MyTopshelfService>(s =>
        {
            s.ConstructUsing(() =>
            {
                var builder = new ContainerBuilder();
                builder.RegisterModule(new AutofacModule());
                var container = builder.Build();

                return container.Resolve<MyTopshelfService>();
            });

            s.WhenStarted((service, control) => service.Start());
            s.WhenStopped((service, control) => service.Stop());
        });

        hostConfigurator.SetDisplayName("MyTopshelfService");
        hostConfigurator.SetServiceName("MyTopshelfService");
        hostConfigurator.SetDescription("My Topshelf Service");
    }

    /// <summary>
    /// This is the entry point of the service host process.
    /// </summary>
    static int Main()
    {
#if DEBUG
        return (int)HostFactory.Run(Configure);
#else 
        try
        {
            ServiceRuntime.RegisterServiceAsync("MyServiceType",
                context =>
                {
                    var builder = new ContainerBuilder();
                    builder.RegisterModule(new AutofacModule());
                    builder.Register(c => new MyService(context, c.Resolve<IMyBusFactory>()))
                        .AsSelf();

                    var container = builder.Build();

                    return container.Resolve<MyService>();

                }).GetAwaiter().GetResult();

            ServiceEventSource.Current.ServiceTypeRegistered(Process.GetCurrentProcess().Id, typeof(MyService).Name);

            // Prevents this host process from terminating so services keep running.
            Thread.Sleep(Timeout.Infinite);
        }
        catch (Exception e)
        {
            ServiceEventSource.Current.ServiceHostInitializationFailed(e.ToString());
            throw;
        }
#endif
        return 0;
    }
}

This is still a work in progress, and I am still tinkering with a few options, but I'll see if anything turns out better.

Partitioning

I haven't had to use partitions yet, but from what I understand the middleware that MT (...GreenPipes) offers is only partitioning
consumers on a single node. The Partitioning that Service Fabric offers is partitioning across different nodes in the cluster. So this
is a whole different ball of wax, and with my minimal experience with it, I'd defer to the experts cough @phatboyg

State

This one is a bit more interesting, and like you said, I think it offers an alternative for MT Persistence. I'm thinking Sagas, assuming
there is an ability to handle transaction serializable, or an optimistic form of concurrency.

[alexeyzimarev]

The Azure docs says:

There are really two types of stateless service solutions. The first one is a service that persists its state externally, for example in an Azure SQL database (like a website that stores the session information and data). The second one is computation-only services (like a calculator or image thumbnailing) that do not manage any persistent state.
In either case, partitioning a stateless service is a very rare scenario--scalability and availability are normally achieved by adding more instances.

So why do you want to partition them? MassTransit is completely stateless, even if you think of sagas - they persist state outside of the running code.

[maldworth]

For partitions, I agree, I cannot think of a scenario, but I don't use them myself, so I could be missing something.

You are correct, Saga states are handled with an external persistent storage mechanism. But a Stateful Service Fabric does offer a Persistent storage mechanism (I'm not sure what mechanism is is leveraging underneath).

If you look at Kevin's example, you can see that it has a StateManager, that supports transactions. So my comment above was implying that this could perhaps be used as a SagaRepository for MassTransit in place of a traditional SQL/noSql external store. Obviously some investigation must be done to verify.

[alexeyzimarev]

Yeah but I think the idea of stateful services if not the best one around. Once one of them fails - everything that is there will stop working.

[pvanroos]

@alexeyzimarev I don't think that is true. If I understand how a Service Fabric cluster works correctly--if one of them fails (the primary replica), an active secondary replica will take over (change role) and pick up where the primary failed, Assuming you stored your state in the reliable collections structure, it can pick up where it left off.

Reference: https://docs.microsoft.com/en-us/azure/service-fabric/service-fabric-reliable-services-lifecycle

[alexeyzimarev]

Picking where you left it is not the same as scaling.
On Thu, 6 Apr 2017 at 22:35, Paul VanRoosendaal [e-mail]
wrote:

@alexeyzimarev https://github.com/alexeyzimarev I don't think that is
true. If I understand how a Service Fabric cluster works correctly--if one
of them fails (the primary replica), an active secondary replica will take
over (change role) and pick up where the primary failed, Assuming you
stored your state in the reliable collections structure, it can pick up
where you left off.

Reference:
https://docs.microsoft.com/en-us/azure/service-fabric/service-fabric-reliable-services-lifecycle

—
You are receiving this because you were mentioned.

Reply to this email directly, view it on GitHub
https://github.com/MassTransit/MassTransit/issues/807#issuecomment-292309754,
or mute the thread
https://github.com/notifications/unsubscribe-auth/ACsMVdZP3Wi_1a6BRfF0SrsO_sC5psomks5rtUyCgaJpZM4MIXLe
.

>

Med vennligst hilsen / Best regards,
Alexey V. Zimarev

[kevbite]

@alexeyzimarev If we stored state for masstransit for like sagas then we need to know which partition stored that data originally so we'd need to be able to route that message somehow to the correct partition. However for most application you could just have 1 partition. thus way we could just get rid of having an external store for state and its fully replicated throughout the cluster.

[pvanroos]

@kevbite

You would route that message to the right partition by using a stateless service (with an http listener) as an intermediary. The stateless service would serve as a sort of broker by examining the contents of the message and sending to the right replica using a partition resolver. The problem is that single stateless service now becomes the bottleneck, but that could be mitigated by distributing multiple instances of the stateless service across the cluster (one per partition) and using a load balancer for incoming traffic to the stateless service (which all of this comes out-of-box for service fabric in Azure).

I think the more challenging part is determining a partition key scheme for the sagas. You could use a composite of some data and hash it, or you could use the distinguishing properties of the message.

The key point to keep in mind is that all of the services that manage and react to the flow of messages in the service bus could be stateful services that are deployed and tied together in a Sf cluster. Communication would be fast and easy between them and any kind of application can publish and subscribe to them since any Sf service can expose any kind of listener or replica on any kind of protocol.

[phatboyg]

I do this on service bus using SessionId. And storing saga state in the session state of service bus.

Does service Fabric have a header that can be used somehow to achieve the same goal?

[pvanroos]

@phatboyg

Sorry for the late reply. I'm still researching this. I looked at the listeners for both stateless and stateful services in Service Fabric (Sf), hoping to find a way to "filter" or "bind" by SessionId or PartitionKey. If I could, it would be possible to create stateful services that "listen" only for certain messages with a particular SessionId or ParitionKey. I couldn't find anything but maybe I'm missing something.

Since I couldn't find a way, I decided to clone the Demo-Registration project and update it by creating stateless brokers between the senders and the StateService.. My specific goals were to replace topshelf-driven services and the Sql server persistent store (EntityFrameworkSagaRepository) with Sf services and the ReliableCollections (i.e. ReliableSagaRepository) persistence store, respectively.

In this approach, the RegistrationService was replaced with a RegistrationStatelessService. The RegistrationActivityService was replaced with a RegistrationActivityStatelessService and most importantly, the StateService was replaced with a RegistrationStateStatefulService. As noted above, the ReliableSagaRepository was written as MassTransit.ReliableCollectionsIntegration for the Persistence folder in MassTransit. Each service uses the MassTransitListener that @kevbite wrote in his library to connect to the bus.

The RegistrationStateStatefulService manages state for the SagaStateMachine using an IReliableDictionary just as it did using a SQL server table. The RegistrationService brokers commands and segments traffic using the first letter of the username in RegistrationStateInstance's. I could have used a different field or scheme to partition RegistrationStateStatefulService (like SessionId or PartitionKey) and it's up to the architect of the system to decide that. 26 partitions were created for each letter in the English alphabet.

Some things to keep in mind: You can define up to 24 instances in a cluster of the RegistrationService to scale horizontally against an Azure service bus that has partitioning enabled for the queues and topics. These 24 instances will connect to the 26 instances of RegistrationStateStatefulService to locate the right partition based on the first letter of the username or whatever you want to use to define the partition scheme. You can get current state by communicating with the RegistrationService as they can expose a web listener much like a WebApi to retrieve state for any given registration. The Sf cluster has a number of system services to manage the replicas for each one of those 26 instances of RegistrationStateStatefulService to ensure high availability and fault tolerance.

The following issue is very important in making the above scenario work using SessionId/PartitionKey: []https://github.com/MassTransit/MassTransit/issues/755

Please let me know what you think. I may be way off base or missing something...

[phatboyg]

Good stuff here, but honestly, with container-based deployments and other things, I doubt SF survives another year. So I'm going to close this.

[pvanroos]

Kubernetes rules! (for the moment) :-)

[kevbite]

Yeah totally agree, Kubernetes seems to be a good path to go down at the moment.

[knowhoper]

ASF has been around over a decade, it powers a huge chunk of Microsofts Azure offerings:

Service Fabric powers many Microsoft services today, including Azure SQL Database, Azure Cosmos DB, Cortana, Microsoft Power BI, Microsoft Intune, Azure Event Hubs, Azure IoT Hub, Dynamics 365, Skype for Business, and many core Azure services.

I don't think it's going away. It would be nice to see an extension to MassTransit that understands SF partitions. But I guess it's a pretty big job.