running dpl in k8s vis ECS

Author: justonedev1

control-operator/api/v1alpha1/environment_types.go

@@ -81,6 +81,7 @@ type EnvironmentStatus struct {
 type TaskReference struct {
 	Name           string            `json:"name"`
 	TaskID         string            `json:"taskID,omitempty"`
+	NameSuffix     string            `json:"nameSuffix,omitempty"`
 	Env            []v1.EnvVar       `json:"env"`
 	ArgsCLI        []string          `json:"argsCLI"`
 	ArgsTransition map[string]string `json:"argsTransition,omitempty"`

control-operator/config/crd/bases/aliecs.alice.cern_environments.yaml

@@ -4100,6 +4100,8 @@ spec:
                         type: array
                       name:
                         type: string
+                      nameSuffix:
+                        type: string
                       taskID:
                         type: string
                     required:

control-operator/ecs-manifests/task-templates/dpl-tasktemplate.yaml

@@ -0,0 +1,62 @@
+apiVersion: aliecs.alice.cern/v1alpha1
+kind: TaskTemplate
+metadata:
+  name: dpl
+  namespace: alice-tasks
+spec:
+  envVars:
+    - O2_DETECTOR
+    - O2_PARTITION
+    - OCC_CONTROL_PORT
+    - O2_SYSTEM
+    - O2_ROLE
+  pod:
+    hostNetwork: true
+    hostIPC: true
+    securityContext:
+      fsGroup: 1100
+      supplementalGroups: [10, 1105]
+    containers:
+      - name: dpl
+        image: gitlab-registry.cern.ch/pkonopka/dockerfiles/flp-sw:20260729-1
+        command: ["bash", "-c"]
+        securityContext:
+          privileged: true
+          runAsUser: 1100
+          runAsGroup: 1100
+        volumeMounts:
+          - name: host-shm
+            mountPath: /dev/shm
+          - name: group
+            mountPath: /etc/group
+            readOnly: true
+          - name: passwd
+            mountPath: /etc/passwd
+            readOnly: true
+          - name: tmp
+            mountPath: /tmp
+          - name: modules
+            mountPath: /lib/modules
+        imagePullPolicy: IfNotPresent
+    volumes:
+      - name: host-shm
+        hostPath:
+          path: /dev/shm
+          type: Directory
+      - name: group
+        hostPath:
+          path: /etc/group
+      - name: passwd
+        hostPath:
+          path: /etc/passwd
+      - name: tmp
+        hostPath:
+          path: /tmp
+      - name: modules
+        hostPath:
+          path: /lib/modules
+    imagePullSecrets:
+      - name: gitlab-registry-secret
+  control:
+    mode: "fairmq"
+    # port: ${OCC_CONTROL_PORT} # to be filled in

control-operator/internal/controller/environment_controller.go

@@ -57,16 +57,27 @@ func (r *EnvironmentReconciler) runTasksFromReferenceOnNode(ctx context.Context,
 	nodename string, resolvedNodename string, req ctrl.Request, environment *aliecsv1alpha1.Environment, log logr.Logger,
 ) (*ctrl.Result, error) {
 	for _, taskReference := range taskReferences {
-		log.Info("geting stored template for task", "task", taskReference.Name)
+
+		log.Info("getting stored template for task", "task", taskReference.Name)
+		taskTemplateName := taskReference.Name
+		if strings.HasPrefix(taskTemplateName, "jit-") {
+			log.Info("getting dpl TaskTemplate for task", "task", taskReference.Name)
+			taskTemplateName = "dpl"
+		}
 		template := &aliecsv1alpha1.TaskTemplate{}
-		if err := r.Get(ctx, types.NamespacedName{Namespace: req.Namespace, Name: taskReference.Name}, template); err != nil {
+		if err := r.Get(ctx, types.NamespacedName{Namespace: req.Namespace, Name: taskTemplateName}, template); err != nil {
 			log.Error(err, "failed to get template for task", "task", taskReference.Name)
 			return &ctrl.Result{}, nil
 		}
 
 		task := &aliecsv1alpha1.Task{}
 		task.Namespace = req.Namespace
 		task.Name = fmt.Sprintf("%s-%s", nodename, template.Name)
+		if taskReference.NameSuffix != "" {
+			task.Name = fmt.Sprintf("%s-%s", task.Name, taskReference.NameSuffix)
+		}
+		task.Name = strings.ToLower(task.Name)
+
 		if err := r.Get(ctx, types.NamespacedName{Name: task.Name, Namespace: task.Namespace}, task); err == nil {
 			continue
 		}

core/config.go

@@ -129,6 +129,7 @@ func setDefaults() error {
 	viper.SetDefault("enableKafka", true)
 	viper.SetDefault("logAllIL", false)
 	viper.SetDefault("metricsEndpoint", "8088/ecsmetrics")
+	viper.SetDefault("jitK8sBasePort", uint16(32000))
 	return nil
 }
 
@@ -200,6 +201,7 @@ func setFlags() error {
 	pflag.Bool("enableKafka", viper.GetBool("enableKafka"), "Turn on the kafka messaging")
 	pflag.Bool("logAllIL", viper.GetBool("logAllIL"), "Send all the logs into IL, including Debug and Trace messages")
 	pflag.String("metricsEndpoint", viper.GetString("metricsEndpoint"), "Http endpoint from which metrics can be scraped: [port/endpoint]")
+	pflag.Uint16("jitK8sBasePort", viper.GetUint16("jitK8sBasePort"), "First control port to allocate for JIT tasks on Kubernetes, incremented per-node")
 
 	pflag.Parse()
 	return viper.BindPFlags(pflag.CommandLine)

core/task/managerk8s.go

@@ -28,6 +28,8 @@ import (
 	"context"
 	"fmt"
 	"maps"
+	"regexp"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -52,8 +54,29 @@ const (
 	k8sDeployTimeout     = 80 * time.Second
 	k8sTransitionTimeout = 80 * time.Second
 	k8sWatchRetryDelay   = 5 * time.Second
+
+	// k8sJitTaskTemplateName is the shared TaskTemplate used to run JIT/DPL
+	// pipeline tasks, since their task class name is unique per generated
+	// workflow and can't be pre-registered as its own TaskTemplate.
+	k8sJitTaskTemplateName = "dpl"
 )
 
+// jitClassNameRe matches JIT-generated task class identifiers of the form
+// "jit-<40-hex-char-sha1>-<devicename>" (see configuration/template/dplutil.go)
+// and captures the devicename, e.g. "readout-proxy" or "Dispatcher".
+var jitClassNameRe = regexp.MustCompile(`^jit-[0-9a-f]{40}-(.+)$`)
+
+func isJitClassName(name string) bool {
+	return jitClassNameRe.MatchString(name)
+}
+
+// jitOnK8sEnabled reports whether ECS may route JIT/DPL task classes to the
+// K8s path. Missing/false means JIT keeps running through Mesos, same as
+// before this bridge existed.
+func jitOnK8sEnabled() bool {
+	return viper.GetBool("jitOnK8s")
+}
+
 // k8sEnvRegistry maps ECS environment IDs to K8s custom Environment names.
 type k8sEnvRegistry struct {
 	mu   sync.RWMutex
@@ -83,6 +106,27 @@ func (r *k8sEnvRegistry) delete(envId uid.ID) {
 	r.mu.Unlock()
 }
 
+// k8sPortAllocator hands out control ports for JIT tasks, one monotonically
+// increasing counter per node (hostNetwork is used, so ports must not collide
+// between JIT tasks scheduled to the same node)
+type k8sPortAllocator struct {
+	portBase uint16
+	next     map[string]uint16
+}
+
+func newK8sPortAllocator(port uint16) *k8sPortAllocator {
+	return &k8sPortAllocator{portBase: port, next: make(map[string]uint16)}
+}
+
+func (allocator *k8sPortAllocator) allocate(hostname string) uint16 {
+	port, ok := allocator.next[hostname]
+	if !ok {
+		port = allocator.portBase
+	}
+	allocator.next[hostname] = port + 1
+	return port
+}
+
 // newK8sClientFromViper creates a K8s client from viper config.
 // Returns nil, nil if kubeNamespace is not configured (K8s disabled).
 func newK8sClientFromViper() (*k8sclient.Client, error) {
@@ -112,7 +156,7 @@ func (m *Manager) deployKubernetesTasks(ctx context.Context, envId uid.ID, descr
 		return nil, err
 	}
 
-	nodeToRefs, err := m.buildK8sNodeTaskRefs(entries)
+	nodeToRefs, err := m.buildK8sNodeTaskRefs(envId, entries)
 	if err != nil {
 		log.WithField("partition", envId).WithError(err).Error("failed to build K8s node task refs")
 		return nil, err
@@ -187,8 +231,12 @@ func (m *Manager) createK8sTaskEntries(envId uid.ID, descriptors Descriptors) ([
 	return entries, nil
 }
 
-func (m *Manager) buildK8sNodeTaskRefs(entries []k8sTaskEntry) (map[string][]v1alpha1.TaskReference, error) {
+const jitK8sBasePortStr = "jitK8sBasePort"
+
+func (m *Manager) buildK8sNodeTaskRefs(envId uid.ID, entries []k8sTaskEntry) (map[string][]v1alpha1.TaskReference, error) {
 	nodeToRefs := make(map[string][]v1alpha1.TaskReference)
+
+	portAllocator := newK8sPortAllocator(viper.GetUint16(jitK8sBasePortStr))
 	for _, e := range entries {
 		t := e.task
 		desc := e.desc
@@ -217,18 +265,50 @@ func (m *Manager) buildK8sNodeTaskRefs(entries []k8sTaskEntry) (map[string][]v1a
 			}
 		}
 
-		argsCLI := make([]string, 0, len(cmd.Arguments))
-		for _, arg := range cmd.Arguments {
-			if strings.TrimSpace(arg) != "" {
-				argsCLI = append(argsCLI, arg)
+		refName := taskClass.Identifier.Name
+		nameSuffix := ""
+		isJit := false
+		if match := jitClassNameRe.FindStringSubmatch(refName); match != nil {
+			isJit = true
+			refName = k8sJitTaskTemplateName
+			nameSuffix = match[1]
+		}
+
+		var argsCLI []string
+		if isJit {
+			// Unlike readout/stfbuilder/stfsender (one fixed port per task type,
+			// baked into their TaskTemplate), JIT devices are dynamic and several
+			// can run on the same node, so each needs its own control port. K8s has
+			// no Mesos-style resource offer to claim a verified-free port from, so
+			// ECS tracks per-node allocation itself (mirrors scheduler.go's
+			// Mesos-offer-based control port claim for FAIRMQ tasks).
+			controlPort := portAllocator.allocate(t.hostname)
+			cmd.Env = append(cmd.Env, fmt.Sprintf("OCC_CONTROL_PORT=%d", controlPort))
+			cmd.Arguments = append(cmd.Arguments, "--control-port", strconv.FormatUint(uint64(controlPort), 10))
+
+			// The "dpl" TaskTemplate runs `bash -c <script>`, so the whole
+			// generated pipeline (driver command + OCC flags, joined the same
+			// way the Mesos executor does it) must be a single Args entry.
+			value := ""
+			if cmd.Value != nil {
+				value = *cmd.Value
+			}
+			argsCLI = []string{strings.Join(append([]string{value}, cmd.Arguments...), " ")}
+		} else {
+			argsCLI = make([]string, 0, len(cmd.Arguments))
+			for _, arg := range cmd.Arguments {
+				if strings.TrimSpace(arg) != "" {
+					argsCLI = append(argsCLI, arg)
+				}
 			}
 		}
 
 		ref := v1alpha1.TaskReference{
-			Name:    taskClass.Identifier.Name,
-			TaskID:  t.taskId,
-			ArgsCLI: argsCLI,
-			Env:     cmdEnvToK8sEnvVars(cmd.Env),
+			Name:       refName,
+			TaskID:     t.taskId,
+			NameSuffix: nameSuffix,
+			ArgsCLI:    argsCLI,
+			Env:        cmdEnvToK8sEnvVars(cmd.Env),
 		}
 		nodeToRefs[t.hostname] = append(nodeToRefs[t.hostname], ref)
 	}