Packages:
Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +“License”); you may not use this file except in compliance +with the License. You may obtain a copy of the License at
+http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +“AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License.
Resource Types:middleManagerDrainStrategyMiddleManagerDrainStrategy enables operator-managed draining before +MiddleManager StatefulSet pods are rolled. If nil, MiddleManagers use the +standard StatefulSet rolling update behavior.
+defaultProbesmiddleManagerDrainmiddleManagerDrainStrategyMiddleManagerDrainStrategy enables operator-managed draining before +MiddleManager StatefulSet pods are rolled. If nil, MiddleManagers use the +standard StatefulSet rolling update behavior.
+defaultProbes+(Appears on: +DruidClusterStatus) +
+MiddleManagerDrainStatus reports an in-progress MiddleManager drain rollout.
+| Field | +Description | +
|---|---|
+statefulSet+ +string + + |
++ | +
+phase+ +string + + |
++ | +
+podName+ +string + + |
++ | +
+podOrdinal+ +int32 + + |
++ | +
+oldPodUID+ +string + + |
++ | +
+lastTransitionTime+ + +Kubernetes meta/v1.Time + + + |
++ | +
+message+ +string + + |
++ | +
+(Appears on: +DruidSpec) +
+MiddleManagerDrainStrategy configures operator-managed draining before a +MiddleManager StatefulSet pod is rolled to a new revision.
+| Field | +Description | +
|---|---|
+drainTimeout+ + +Kubernetes meta/v1.Duration + + + |
+
+(Optional)
+ DrainTimeout is the maximum time to wait for streaming ingestion tasks to +drain before allowing Kubernetes to replace the MiddleManager pod. + |
+
+podReadyTimeout+ + +Kubernetes meta/v1.Duration + + + |
+
+(Optional)
+ PodReadyTimeout is the maximum time to wait for Kubernetes to replace the +pod and for the replacement to become ready on the target StatefulSet revision. + |
+
diff --git a/docs/druid_cr.md b/docs/druid_cr.md index a082062..3184e13 100644 --- a/docs/druid_cr.md +++ b/docs/druid_cr.md @@ -31,6 +31,11 @@ spec: # more information in features.md and in druid documentation # http://druid.io/docs/latest/operations/rolling-updates.html rollingDeploy: true + # Optional: drain MiddleManager StatefulSet pods before replacing them during rolling updates. + # Omit this field to use standard StatefulSet rolling update behavior. + middleManagerDrainStrategy: + drainTimeout: 1h + podReadyTimeout: 30m # Image for druid, Required Key image: apache/druid:25.0.0 .... diff --git a/docs/features.md b/docs/features.md index 6f8b74b..2eab033 100644 --- a/docs/features.md +++ b/docs/features.md @@ -23,6 +23,7 @@ under the License. - [Finalizer in Druid CR](#finalizer-in-druid-cr) - [Deletion of Orphan PVCs](#deletion-of-orphan-pvcs) - [Rolling Deploy](#rolling-deploy) +- [MiddleManager Drain Strategy](#middlemanager-drain-strategy) - [Force Delete of Sts Pods](#force-delete-of-sts-pods) - [Horizontal Scaling of Druid Pods](#horizontal-scaling-of-druid-pods) - [Volume Expansion of Druid Pods Running As StatefulSets](#volume-expansion-of-druid-pods-running-as-statefulsets) @@ -68,6 +69,39 @@ Default updates are done in parallel. Since cluster creation does not require a in parallel anyway. To enable this feature, set `rollingDeploy: true` in the Druid CR. ⚠️ This feature is enabled by default. +## MiddleManager Drain Strategy +For streaming ingestion clusters, a regular StatefulSet rolling update can terminate MiddleManager pods while they still +own running Kafka or Kinesis indexing tasks. The operator can instead drain each MiddleManager pod before Kubernetes +replaces it. + +Enable this opt-in behavior by setting `spec.middleManagerDrainStrategy`. The field uses pointer presence: if the field +is omitted, MiddleManagers use the standard StatefulSet rolling update behavior. + +```yaml +spec: + rollingDeploy: true + middleManagerDrainStrategy: + drainTimeout: 1h + podReadyTimeout: 30m +``` + +When enabled, the operator: + +1. Blocks the MiddleManager StatefulSet rollout by setting the rolling update partition to the replica count. +2. Selects one outdated MiddleManager pod at a time, highest ordinal first. +3. Disables the worker through the Druid Overlord API. +4. Finds running streaming tasks with Druid SQL and triggers supervisor task-group handoff. +5. Waits for running Kafka/Kinesis tasks on that pod to drain, up to `drainTimeout`. +6. Lowers the StatefulSet partition for that pod and waits for the replacement pod to be ready on the new revision. +7. Re-enables the worker and proceeds to the next pod on the following reconcile. + +The strategy is supported only for MiddleManager nodes running as StatefulSets. If configured for a MiddleManager +Deployment, the operator logs a warning and uses the normal Deployment rollout path. + +This feature requires the operator to reach the Druid Router service and use the Druid API credentials configured in +`spec.auth`, when authentication is enabled. If the Router service cannot be discovered, the rollout is blocked and the +error is surfaced by the reconcile loop. + ## Force Delete of Sts Pods During upgradeS, if THE StatefulSet is set to `OrderedReady` - the StatefulSet controller will not recover from crash-loopback state. The issues is referenced [here](https://github.com/kubernetes/kubernetes/issues/67250). diff --git a/pkg/druidapi/druidapi.go b/pkg/druidapi/druidapi.go index 3cf4749..028f87c 100644 --- a/pkg/druidapi/druidapi.go +++ b/pkg/druidapi/druidapi.go @@ -141,6 +141,18 @@ func MakePath(baseURL, componentType, apiType string, additionalPaths ...string) return u.String() } +// MakeSQLPath constructs the path for Druid's SQL API. +func MakeSQLPath(baseURL string) string { + u, err := url.Parse(baseURL) + if err != nil { + fmt.Println("Error parsing URL:", err) + return "" + } + + u.Path = path.Join("druid", "v2", "sql") + return u.String() +} + // GetRouterSvcUrl retrieves the URL of the Druid router service. // Parameters: // diff --git a/pkg/druidapi/druidapi_test.go b/pkg/druidapi/druidapi_test.go index ebcb6db..b7d80e6 100644 --- a/pkg/druidapi/druidapi_test.go +++ b/pkg/druidapi/druidapi_test.go @@ -182,3 +182,36 @@ func TestMakePath(t *testing.T) { }) } } + +func TestMakeSQLPath(t *testing.T) { + tests := []struct { + name string + baseURL string + expected string + }{ + { + name: "RouterService", + baseURL: "http://example-druid-service", + expected: "http://example-druid-service/druid/v2/sql", + }, + { + name: "BaseURLWithPath", + baseURL: "http://example-druid-service/base", + expected: "http://example-druid-service/druid/v2/sql", + }, + { + name: "EmptyBaseURL", + baseURL: "", + expected: "druid/v2/sql", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + actual := MakeSQLPath(tt.baseURL) + if actual != tt.expected { + t.Errorf("MakeSQLPath() = %v, expected %v", actual, tt.expected) + } + }) + } +}