How to scale the pipeline elements service?

[RobertIndie]

Hi all,

Apache StreamPipes as a IoT data stream processing system, is necessary to provide robust computational capabilities. The pipeline element service, which primarily handles data stream processing, may require scalable support to optimize its performance. I would like to initiate a discussion to discuss about how to scale the pipeline element service.

Here are some key requirements that I believe we should address:

• Replica: The pipeline element service should have the ability to set up replicas, enabling us to leverage Kubernetes (k8s) for automatic scaling.
• Routing: Each element could be routed to one element service instance(a k8s replica)
• Load balancing: Support load balancing for elements. If one instance is under high load, support for transferring part of the elements to another instance
• Failover: If one instance failed(which could be detected by the k8s), other instances should be able to swiftly take over and restore the running elements from the failed instance.

I would like to discuss how we achieve these needs. Do you know if there are any libraries or middleware that could facilitate this? Do you have any ideas? @dominikriemer @tenthe @bossenti @luoluoyuyu

[dominikriemer]

Hi @RobertIndie thanks for starting this discussion!
It should be already possible to start multiple instances of a pipeline element service, but currently, the core would just randomly choose a service when invoking pipeline elements. I think it would be a very nice feature to properly place pipeline elements based on load and also to transfer pipeline elements in case of failures. The current health monitoring component should be able to detect a failed service and restart the pipeline element in case there is another supported service running. One thing we currently do not support, but what would be important to address in this context is state management of stateful pipeline elements, so that state is properly restored when the pipeline element is reinvoked.

We had a community member who did some cool initial work on this (https://github.com/apache/streampipes/tree/edge-extensions) but unfortunately it never made it into our development branch.

I think one good thing to get started is the replica feature - configure our helm charts to set up replicas and to check if the current health management properly recovers the pipeline elements? And we can also start to discuss how to collect load metrics - changing the core to select a service based on load shouldn't be too hard.

It would be definitely great to have these features!

[RobertIndie]

Thanks for your sharing. I'll continue to investigate this logic. And will share my further thoughts here.

[bossenti]

Hi @RobertIndie,

I was quite unsure how to express my thoughts here. In general, I think it's great that you brought this up @RobertIndie. Thanks a lot 🙏🏼
However, it's hard for me to come up with something more concrete to say at this abstract level.
But I like @dominikriemer's idea to start with the replica feature and tackle the other parts iteratively.

Looking forward to the next steps 🤩

[RobertIndie]

Thanks for your helpful input! I have checked the code, but I still have the following questions.

• Is the selection of the extensions service endpoint truly random? From the code, it seems it only selects the first one. Is the order of the list randomized? When restoring the pipeline element, it seems always to select the first endpoint.
• Regarding detecting the failed element instance, it seems that we could only check the failed instance at every health check(60s). But isn't it sometimes too late to perceive the failed instance? Am I missing some logic?

[dominikriemer]

Good questions!

• Currently, indeed the first matching endpoint is selected. I think it would be not too difficult to change that to a random selection strategy. Afterwards, we could also extend the monitoring component so that usage metrics are sent to the core and could implement a selection strategy which takes resource utilization into account.
• Yes, the health check currently runs every 60 seconds. Would you prefer to reduce the interval? What would be a good interval so that we don't iss too much data and do you think a very short interval would be ok in terms of the number of requests we would send to the extensions for checking availability?

[luoluoyuyu]

Hi all
I'm looking to add a load balancer for Streampipes. Implement the above-mentioned routing, load balancing, failover, and other functions through the load balancer. and launched a proposal (#2552)