[Feature] Parallel Executor

CHANGELOG.md

@@ -3,6 +3,7 @@
 ## [master](https://github.com/arangodb/kube-arangodb/tree/master) (N/A)
 - (Feature) (Platform) MetaV1 Integration Service
 - (Feature) (Platform) Chart Overrides
+- (Feature) Parallel Executor
 
 ## [1.2.49](https://github.com/arangodb/kube-arangodb/tree/1.2.49) (2025-06-17)
 - (Maintenance) Optimize go.mod

pkg/deployment/resources/pod_inspector.go

@@ -35,6 +35,7 @@ import (
 	"github.com/arangodb/kube-arangodb/pkg/deployment/agency/state"
 	"github.com/arangodb/kube-arangodb/pkg/deployment/features"
 	"github.com/arangodb/kube-arangodb/pkg/deployment/patch"
+	"github.com/arangodb/kube-arangodb/pkg/handlers/utils"
 	"github.com/arangodb/kube-arangodb/pkg/metrics"
 	"github.com/arangodb/kube-arangodb/pkg/util"
 	"github.com/arangodb/kube-arangodb/pkg/util/constants"
@@ -106,7 +107,7 @@ func (r *Resources) InspectPods(ctx context.Context, cachedStatus inspectorInter
 
 		spec := r.context.GetSpec()
 		groupSpec := spec.GetServerGroupSpec(group)
-		coreContainers := spec.GetCoreContainers(group)
+		coreContainers := getPodCoreContainers(spec.GetCoreContainers(group), pod.Spec.Containers)
 
 		if c, ok := memberStatus.Conditions.Get(api.ConditionTypeUpdating); ok {
 			if v, ok := c.Params[api.ConditionParamContainerUpdatingName]; ok {
@@ -523,3 +524,15 @@ func removeLabel(labels map[string]string, key string) map[string]string {
 
 	return labels
 }
+
+func getPodCoreContainers(coreContainers utils.StringList, containers []core.Container) utils.StringList {
+	r := make(utils.StringList, 0, len(coreContainers))
+
+	for _, container := range containers {
+		if coreContainers.Has(container.Name) {
+			r = append(r, container.Name)
+		}
+	}
+
+	return r
+}

pkg/util/executor/executor.go

@@ -0,0 +1,156 @@
+//
+// DISCLAIMER
+//
+// Copyright 2025 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed 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.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package executor
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/rs/zerolog"
+
+	"github.com/arangodb/kube-arangodb/pkg/logging"
+	"github.com/arangodb/kube-arangodb/pkg/util/errors"
+)
+
+func Run(ctx context.Context, log logging.Logger, threads int, f RunFunc) error {
+	h := &handler{
+		th:        NewThreadManager(threads),
+		completed: make(chan struct{}),
+		log:       log,
+	}
+
+	go h.run(ctx, f)
+
+	return h.Wait()
+}
+
+type RunFunc func(ctx context.Context, log logging.Logger, t Thread, h Handler) error
+
+type Executor interface {
+	Completed() bool
+
+	Wait() error
+}
+
+type Handler interface {
+	RunAsync(ctx context.Context, f RunFunc) Executor
+
+	WaitForSubThreads(t Thread)
+}
+
+type handler struct {
+	lock sync.Mutex
+
+	th ThreadManager
+
+	handlers []*handler
+
+	completed chan struct{}
+
+	log logging.Logger
+
+	err error
+}
+
+func (h *handler) WaitForSubThreads(t Thread) {
+	for {
+		t.Release()
+
+		if h.subThreadsCompleted() {
+			return
+		}
+
+		time.Sleep(10 * time.Millisecond)
+	}
+}
+
+func (h *handler) subThreadsCompleted() bool {
+	h.lock.Lock()
+	defer h.lock.Unlock()
+
+	for id := range h.handlers {
+		if !h.handlers[id].Completed() {
+			return false
+		}
+	}
+
+	return true
+}
+
+func (h *handler) Wait() error {
+	<-h.completed
+
+	return h.err
+}
+
+func (h *handler) Completed() bool {
+	select {
+	case <-h.completed:
+		return true
+	default:
+		return false
+	}
+}
+
+func (h *handler) RunAsync(ctx context.Context, f RunFunc) Executor {
+	h.lock.Lock()
+	defer h.lock.Unlock()
+
+	n := &handler{
+		th:        h.th,
+		completed: make(chan struct{}),
+		log:       h.log,
+	}
+
+	h.handlers = append(h.handlers, n)
+
+	go n.run(ctx, f)
+
+	return n
+}
+
+func (h *handler) run(ctx context.Context, entry RunFunc) {
+	defer close(h.completed)
+
+	err := h.runE(ctx, entry)
+
+	subErrors := make([]error, len(h.handlers))
+
+	for id := range subErrors {
+		subErrors[id] = h.handlers[id].Wait()
+	}
+
+	subError := errors.Errors(subErrors...)
+
+	h.err = errors.Errors(err, subError)
+}
+
+func (h *handler) runE(ctx context.Context, entry RunFunc) error {
+	t := h.th.Acquire()
+	defer t.Release()
+
+	log := h.log.Wrap(func(in *zerolog.Event) *zerolog.Event {
+		return in.Int("thread", int(t.ID()))
+	})
+
+	return entry(ctx, log, t, h)
+}

pkg/util/executor/executor_test.go

@@ -0,0 +1,59 @@
+//
+// DISCLAIMER
+//
+// Copyright 2025 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed 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.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package executor
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/arangodb/kube-arangodb/pkg/logging"
+)
+
+func Test_Executor(t *testing.T) {
+	ctx := context.Background()
+
+	logger := logging.Global().RegisterAndGetLogger("test", logging.Trace)
+
+	require.NoError(t, Run(ctx, logger, 1, func(ctx context.Context, log logging.Logger, th Thread, h Handler) error {
+		log.Info("Start main thread")
+		defer log.Info("Complete main thread")
+
+		h.RunAsync(ctx, func(ctx context.Context, log logging.Logger, th Thread, h Handler) error {
+			log.Info("Start second thread")
+			defer log.Info("Complete second thread")
+
+			h.RunAsync(ctx, func(ctx context.Context, log logging.Logger, th Thread, h Handler) error {
+				log.Info("Start third thread")
+				defer log.Info("Complete third thread")
+
+				return nil
+			})
+
+			return nil
+		})
+
+		h.WaitForSubThreads(th)
+
+		return nil
+	}))
+}

pkg/util/executor/threader.go

@@ -0,0 +1,97 @@
+//
+// DISCLAIMER
+//
+// Copyright 2025 ArangoDB GmbH, Cologne, Germany
+//
+// Licensed 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.
+//
+// Copyright holder is ArangoDB GmbH, Cologne, Germany
+//
+
+package executor
+
+import "sync"
+
+func NewThreadManager(threads int) ThreadManager {
+	r := make(chan ThreadID, threads)
+
+	for id := 0; id < threads; id++ {
+		r <- ThreadID(id)
+	}
+
+	return &threadManager{
+		threads: r,
+	}
+}
+
+type ThreadID int
+
+type ThreadManager interface {
+	Acquire() Thread
+}
+
+type threadManager struct {
+	threads chan ThreadID
+}
+
+func (t *threadManager) Acquire() Thread {
+	id := <-t.threads
+
+	return &thread{
+		parent: t,
+		id:     id,
+	}
+}
+
+type Thread interface {
+	ID() ThreadID
+
+	Release()
+	Wait()
+}
+
+type thread struct {
+	lock sync.Mutex
+
+	parent *threadManager
+
+	released bool
+
+	id ThreadID
+}
+
+func (t *thread) ID() ThreadID {
+	return t.id
+}
+
+func (t *thread) Release() {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+
+	if t.released {
+		return
+	}
+
+	t.released = true
+
+	t.parent.threads <- t.id
+}
+
+func (t *thread) Wait() {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+
+	t.parent.threads <- t.id
+
+	t.id = <-t.parent.threads
+}