Druid: KubernetesTaskRunner for running druid tasks as kubernetes jobs

[nishantmonu51]

Motivation

Below talk by @Jinchul81 outlines a way to autoscale druid Middlemanagers by emitting druid metrics -
https://www.slideshare.net/Hadoop_Summit/apache-druid-auto-scaleoutin-for-streaming-data-ingestion-on-kubernetes

However, there are some limitations with this approach especially for selecting MMs for scaling down discussed towards end slides.

• A middlemanager can only be scaled down when all its tasks have been completed
• Selecting a specific MM to scale down is not trivial

Another requirement with MMs is over-provisioning where required workerCapacity = 2 * replicas * taskCount to accomodate one set of tasks publishing while another set is reading

Proposed changes

This proposal is to implement a KubernetesTaskRunner as part of a new extension druid-kubernetes

• K8sTaskRunner will use K8s API to directly submit jobs to kubernetes cluster using java kubernetes-client https://github.com/kubernetes-client/java
• Each K8s Job will only run Peon process.
• Kubernetes task runner will use Watch (link) to watch for the status changes of the submitted jobs
• Task logs can will also be streamed in the console using existing kubernetes APIs. (link)
• For replica tasks, kubernetes antiaffinity can be used to make sure that replica tasks are assigned to different instances
• Once the job completes, allocated resources will be freed and can be assigned to other tasks.
• Instead of MM pushing the task logs on completion to log storage, Overlord will fetch the logs and push them to the log storage

Rationale

• KubernetesTaskRunner would help in better resource utilization in the cloud as we need not allocate larger pods which can host multiple tasks that are mostly under-utilized
• Kubernetes ClusterAutoscaler can be used to add more nodes whenever needed, If enough resources are not available, tasks will remain in pending state, cluster-autoscaler will create more nodes for running these tasks
• No over-provisioning is needed to allow simultaneously publishing and reading tasks is required

Operational impact

• Simplified autoscaling in cloud.
• No need to manage extra configuration for MiddleManagers in the cloud
• This is proposed to be done as a kubernetes extension with the hope of adding better cloud support for druid. It will be an Optional opt-in feature and no change in core druid are required.

[jihoonson]

Sounds good to me overall.

Instead of MM pushing the task logs on completion to log storage, Overlord will fetch the logs and push them to the log storage

Could this potentially become a bottleneck when scaling?

[nishantmonu51]

Sounds good to me overall.

Instead of MM pushing the task logs on completion to log storage, Overlord will fetch the logs and push them to the log storage

Could this potentially become a bottleneck when scaling?

This will be done asynchronously by a background thread.

However, if it becomes a bottleneck, log publishing can be turned off and any of kubernetes logging methods can be used to collect job logs -
recommendations are described here - https://kubernetes.io/docs/concepts/cluster-administration/logging/

[himanshug]

+1 , we were actually thinking along those lines in recent past.

[sascha-coenen]

we are also very interested in this kind of stuff and spent quite some time investigating good approaches.

With regard to the current limitations of scaling down in Kubernetes:

• the respective SIG (special interest group) was considering different approaches for how to handle this in future K8s versions. The current idea seems to be that there will be a property named "evictionCost" within a pod's status object. This will help feeding back preferences about which pod of a deployment to kill for a scale-down operation to kubernetes
• the current workaround that is known to the community which is the least hacky indeed seems to be to use Kubernetes Jobs.

With regard to the autoscaling mechanism in Druid to be tied to AWS EC2:

• we wrote an autoscaler for Druid which delegates the execution of provision/terminate/ip2idLookup/id2ipLookup to an external REST endpoint. This is more lightweight than having the platform-specific code sitting inside of a Druid extension which makes it slow/difficult to make quick adjustments. We are willing to contribute this.
• we are currently working towards another contribution that improves the autoscaling mechanism in Druid to support different tiers of middlemanagers. ( #8695 )
  This feature proposal will soon receive a code contribution and we will also update the description of the proposal. It seems to me that in general, one should either make sure that the approach in the scope of this proposal should either be aligned with the existing concept of the autoscaling extension-point in Druid or the current autoscaling mechanism in Druid should be clearly marked as deprecated so that the community can align well on working on future-safe contributions.

Open points I did not think too much about yet:

• how would volume management work in the case of many autoscaled peon/indexer pods
• how to design this approach such that it integrates well with the peer project that works towards a Druid operator for k8s
• how to handle task failover in cases Kubernetes relocates pods to free up physical nodes. Such compactions happen and statefulsets/deployments deal well with such planned operational disruptions via PodDisruptionBudgets and via the internal controller logic of deployments/statefulsets. I don't recall the specifics of k8s Jobs in this context, but it seems that one needs to think explicitly about how to handle such cases. Jobs don't get restarted indefinately often in response to a pod relocation and task failover would need to be supported for native batch tasks such that such planned disruptions would not make a whole multi-task batch ingestion task fail.

By the way: would it make sense if this proposal were to target only Druid's new ingestion pipeline consisting of the Druid indexer and native ingestion tasks such as index_parallel etc. or would it be required to support different modes like middlemanager mode vs. indexer mode.
I would certainly hope that the new native indexing mechanism would be supported along with the new indexer. Not sure though whether it is necessary to also support the middlemanager/peon lane as well. What do you think?

[AdheipSingh]

Talking only from a kubernetes and druid operator perspective in order to solve the problem of scaling MM dynamically.

Scenario:

• To scale MM when a task is in pending state, Scale down MM when their is no task associated with it.

Problems:

• When druid operator scales down a MM we need to make sure we don't scale down a MM pod which is running a task. How can we protect a MM during scale down ?
• Here to use Statefulsets for MM is another blocker. Lets say i have two MM running with statefulsets MM-0 and MM-1. I have a scenario where MM-0 is not running any task where as MM-1 is running a task, in case i scale down kubernetes will delete MM-1, since StatefulSet controllers always removes the most recently created MM.
• With this we can support deployments for MM https://github.com/druid-io/druid-operator/pull/52 .

• Another blocker here is we cannot have a single service encapsulating MM pods. We need to have separate service for each MM. In that way i can have a total distinction of MM pod. So that if i hit this endpoint /druid/worker/v1/tasks for MM-0 and MM-1 i get exact counts for each, with a single service encapsulating it, it will LB between the two pods and give a count for only one of them. So the operator for MM should not have replicas but a count, so in case i specify 2 counts of MM the operator shall deploy 2 deployments of MM with 2 separate services.

• The operator on each reconcile, can hit this endpoint for each MM /druid/worker/v1/tasks and then scale down the particular deployment of MM ( meaning scale down rs to 0, not to delete the MM ). Scale up can be done using v1/indexer/pendingTask endpoint to increase the count of MM.

• On volume management, addressed on the above comment, in case your storage class has reclaimPolicy enabled to true, the PVC shall not be deleted in that case. As per my understanding its fine to have n number MM associated with a common pvc volumeMount. As per current state of the operator, i guess deployments need to be enhanced to support pvc.

These problems have been associated with kafka operators too, and some of the operators have addressed these blockers when using statefulsets on k8s.

HPA cant solve this problem even with external metrics, have tried those approaches. HPA can only work if its supports to scale up on different metrics and scale down on different metrics.

Again just to be clear here i am assuming nothing changes in druid :) commenting from k8s controllers perspective.

@himanshug @nishantmonu51 would like to know your perspective on this approach .

[himanshug]

Original proposal here, to have a K8sTaskRunner, that creates one k8s pod(or job with replica count = 1) per peon entirely removes MMs from the equation. Then "autoscaling" essentially becomes responsibility of K8S scheduler itself. So, that is all great and beneficial for a good class of users.
Some users have custom lookup implementations for large lookups data sizes, wherein lookup data is loaded at the MM level and shared by all the peons on that MM which alleviates a lot of resource wastage. For users of such or other similar features that somehow are benefitted by "sharing" of resources across all Peons running on same MM, above K8sTaskRunner would not be the best option for autoscale probably unless we could still allow the sharing by using EBS etc even in this world.

So, even after K8sTaskRunner (and also easier to implement), it would still be beneficial to some users to have an auto scaling solution that continues to have MMs. Now, there are multiple options of achieving that.

1. Use combination of k8s HPA and custom metrics to achieve that MM autoscaling requirements. From previous comment, it sounds like that HPA's mechanisms are not sophisticated enough at this time to support our requirements. So, this one will be a good choice at some point in the far future when k8s community improves things around HPA.

2. Let Druid Operator manage the autoscaling totally transparently. That is what is proposed in

   The operator on each reconcile, can hit this endpoint for each MM /druid/worker/v1/tasks and then scale down the particular deployment of MM ( meaning scale down rs to 0, not to delete the MM ). Scale up can be done using v1/indexer/pendingTask endpoint to increase the count of MM.

Personally, I think Druid's Overlord process has best access to all state necessary for making auto scaling related decisions, already implements all(and more) of above logic and delegates just the provisioning part to AutoScaler interface . So, in my mind, it would be better to write a K8sAutoScaler to manage MM autoscaling.

From implementation standpoint, one crucial problem is "telling k8s to scale down by killing a particular MM pod" since k8s Deployment/StatefulSet resources can't be told which pod[s] to kill while scaling down. One workaround is, of course, to create one Deployment or StatefulSet resource for each MM individually. This workaround is good because this doesn't need anything non-standard. However, Deployment or StatefulSet etc with replica count 1 don't play well with things like PodDisruptionBudget.

Additionally, another workaround, I would also investigate to see if we can use CloneSet which has the scale down feature that we need. At some point, I think, Deployment would support that feature and we can switch to standard Deployment instead of CloneSet then. Obvious downside is that k8s cluster then needs the controller for CloneSet which isn't automatically available in a vanilla k8s cluster installation. With CloneSet we can have druid-operator deploy the CloneSet resource and K8sAutoScaler would just update the replica counts to scale up/down. In that world, druid-operator would continue to be source of truth about all of deployment except only the MM replica count.

[AWaterColorPen]

@himanshug

Use combination of k8s HPA and custom metrics to achieve that MM autoscaling requirements. From previous comment, it sounds like that HPA's mechanisms are not sophisticated enough at this time to support our requirements. So, this one will be a good choice at some point in the far future when k8s community improves things around HPA.

Is there any possible for MM providing a metrics just like allow-scale-in?
The custom metrics logic and HPA's mechanisms can implement at prometheus-druid-exporter

[AdheipSingh]

@AWaterColorPen

When statefulset controller scales down a MM we need to make sure we don't scale down a MM pod which is running a task. How can we protect a MM during scale down ?

Here` to use Statefulsets for MM is another blocker. Lets say i have two MM running with statefulsets MM-0 and MM-1. I have a scenario where MM-0 is not running any task where as MM-1 is running a task, in case i scale down kubernetes will delete MM-1, since StatefulSet controllers always removes the most recently created MM.

^^ this is a blocker regardless of you using HPA's whatever feature, with custom logic, since HPA will also delegate to deployment controller and statefulset controller as it bound to use that spec. HPA just maintains the replica count.

The only righteous way as @himanshug has mentioned is to use cloneset , which works fine since it does not use deployment or statefullset controller to run pods, each MM pod is treated as unique.

Another blocker here is we cannot have a single service encapsulating MM pods. We need to have separate service for each MM. In that way i can have a total distinction of MM pod. So that if i hit this endpoint /druid/worker/v1/tasks for MM-0 and MM-1 i get exact counts for each, with a single service encapsulating it, it will LB between the two pods and give a count for only one of them. So the operator for MM should not have replicas but a count, so in case i specify 2 counts of MM the operator shall deploy 2 deployments of MM with 2 separate services.

Scale up is fine with HPA on pending Tasks count but scale down can only and only be achieved if you treat each MM as unique which means you will not wrap a MM under a single service, but for each MM-0, MM-1 you will have a seperate service for each.
Plus supporting Mem/CPU for MM does not make any sense to me, Mem/CPU should be only used with brokers nodetype in my knowledge

[AWaterColorPen]

I agree with that statefulset can't solve the MM-0 and MM-1 problem.

How about following problem?
example: we have some MM with druid.worker.capacity = 5

1. At starting, 100 tasks per unit time comes at the same time. MM scale up to 20, running 5 tasks per MM node. It works well.
2. Then, 100 tasks reduce to 30 tasks per unit time for some reason. we expect that 6 MM dealing with the 30 tasks, running 5 tasks per MM node. But druid coordinator/overload won't only schedule 30 tasks on 6 MM. For my understanding, it will schedule 30 tasks at 20 MM on average, running 1-2 tasks per MM node.
   SO THAT, NO MM node can be scale down. It wastes 70% resources. Util no enough tasks scheduled on each MM, it will scale in.

Is the feature of scaling-in what we want?

[himanshug]

Then, 100 tasks reduce to 30 tasks per unit time for some reason. we expect that 6 MM dealing with the 30 tasks, running 5 tasks per MM node. But druid coordinator/overload won't only schedule 30 tasks on 6 MM. For my understanding, it will schedule 30 tasks at 20 MM on average, running 1-2 tasks per MM node.

I think that is easily fixable by changing the selectStrategy to fillCapacity instead of default equalDistribution , please see https://druid.apache.org/docs/latest/configuration/index.html#overlord-dynamic-configuration

[AWaterColorPen]

I think that is easily fixable by changing the selectStrategy to fillCapacity instead of default equalDistribution , please see https://druid.apache.org/docs/latest/configuration/index.html#overlord-dynamic-configuration

Oh, It seems that you use fillCapacity and CloneSet to solve the autoscale problem.
Emmm if the selectStrategy has a feature similar with fillCapacity that it assign task to the worker with lower pod id,
StatefulSet would not be much worse than CloneSet.

For helm chart case, there is few charts using CloneSet. It is not easy to apply CloneSet solution.

[AdheipSingh]

using cloneset directly will not work either, their needs to be custom code written, either run it as a job or hook it into operator code. You will need to use cloneset client and keep the operator + druid + MM in sync since with introducing cloneset, its the responsibility of cloneset for reconciling MM, not the operator.
Here's a simple code https://gist.github.com/AdheipSingh/db009db6122dabd763fa70e0b1d9385b which i was using to experiment between these, i was running this as a job every 5 sec, this would get the pendingTasks from druid and add +1 MM for every pending task...... for scale down it would hit every mm endpoint to check if its running task if not it would delete the svc + cloneset.

Ideally this problem should be definitely solved from druid end, but at the same time its possible to get it working from k8s side.

[zhangyue19921010]

Hi everyone and Hi @nishantmonu51 I happened to propose a similar plan recently. https://github.com/apache/druid/issues/10824 :) Sorry I didn't notice there is a same Issue have been proposed 16 months ago ...

Is there any progress on it so far? If yes, I will close my Issue immediately. If not maybe I can have a try and contribute my work.

What I have achieved is that middlemanager(statefulset) creates peon pod and controls its lifecycle. Peon task is successful, logged and can be queried.

Also Combined with https://github.com/apache/druid/pull/10524. Druid has the ability which is tested to create peon pods and auto scale the pod numbers!

If there are any concerns please let me know.
Looking forward to your reply. Thanks a lot.

[nishantmonu51]

Hi @zhangyue19921010 : I started working on it but then got side-tracked with other issues.
Would be happy to collaborate on this and work with you to get this finalized.
I reached a state where i was able to spin up peons directly from the coordinator/overlord and persist the indexing logs. Still some concurrency issues need to be looked into and testing is pending.

[zhangyue19921010]

Sure. It’s my honor to collaborate on this and work with you to get this finalized. I will close my Issue. Could you please make a PR based on your current state at your convenience so that I could take a look and help you to make it happen because I have finished and tested middlemanger on k8s based on my though before.

Or I also glad to make a pr raising what I have done like I mentioned above although UT-related work is still in progress.

What’s your opinion?
Looking forward to your reply. Thanks

[nishantmonu51]

@zhangyue19921010 Sure, Please raise a PR and we can collaborate on it.

[jasonk000]

~Ideally the K8sTaskRunner becomes an extension point where we can plug in any sort of TaskRunner extension. Is this the intended approach?~

~I think there would be value in looking at this for a different job scheduler implementations (non-k8s, similar in some ideas). Having this as a standardised extension point will allow us to run this on any scheduler.~

~With regards to comments on HPA / scaling requirements, currently we set task properties in Druid ingestion spec incl context.druid.indexer.runner.javaOpts to set thread counts and offheap memory limits, etc. I think we can use a similar mechanism in the ingestion spec to pass specific configuration to the task runner such as the expected resources required, with some defaults provided in configuration. This keeps the spec simple and can be modified for each use case.~

Edit:
I discovered this work has been followed up over here. I think this is a model I could adapt to a custom task scheduler implementation!