run dpl in k8s from ECS

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/kubernetes-manifests/dpl-qccheck-test.yaml

@@ -0,0 +1,110 @@
+apiVersion: aliecs.alice.cern/v1alpha1
+kind: Task
+metadata:
+  labels:
+    environment: 34xmvqdqcsu
+    node: mtichak-ost
+    taskID: 34XmVrU4GeE
+  name: mtichak-ost-dpl-qc-check-daq-qccheck
+  namespace: alice-tasks
+spec:
+  control:
+    mode: fairmq
+    port: 31000
+  nodeName: mtichak-ost.cern.ch
+  pod:
+    containers:
+    - args:
+      - source /etc/profile.d/o2.sh &&  o2-dpl-raw-proxy -b --session default --dataspec
+        'x:TST/RAWDATA;dd:FLP/DISTSUBTIMEFRAME/0' --inject-missing-data --readout-proxy
+        '--channel-config "name=readout-proxy,type=pull,method=connect,address=ipc:///tmp/stf-builder-dpl-pipe-0,transport=shmem,rateLogging=10"'
+        | o2-qc -b --config apricot://mtichak-ost.cern.ch:32188/o2/components/qc/ANY/any/stfb_to_daqtask-mtichak-ost
+        | o2-dpl-output-proxy --environment DPL_OUTPUT_PROXY_ORDERED=1 -b --session
+        default --dataspec 'x:TST/RAWDATA;dd:FLP/DISTSUBTIMEFRAME/0' --dpl-output-proxy
+        '--channel-config "name=downstream,type=push,method=bind,address=ipc:///tmp/stf-pipe-0,rateLogging=10,transport=shmem"'
+        | o2-qc -b --exit-transition-timeout '40' --data-processing-timeout '20' --monitoring-backend
+        'influxdb-unix:///tmp/telegraf-dpl.sock' --session 'default' --infologger-severity
+        'info' --infologger-mode 'infoLoggerD' --driver-client-backend 'stdout://'
+        --shm-segment-size '10000000000' --shm-throw-bad-alloc 'false' --resources-monitoring
+        '15' --id 'qc-check-DAQ-QcCheck' --shm-monitor 'false' --log-color 'false'
+        --batch --aod-file '' --aod-memory-rate-limit '0' --aod-writer-json '' --aod-writer-keep
+        '' --aod-writer-maxfilesize '0' --aod-writer-ntfmerge -1 --aod-writer-resdir
+        '' --aod-writer-resfile '' --aod-writer-resmode 'RECREATE' --bad-alloc-attempt-interval
+        '50' --bad-alloc-max-attempts '1' --channel-prefix '' --clone '' --config
+        'apricot://mtichak-ost.cern.ch:32188/o2/components/qc/ANY/any/stfb_to_daqtask-mtichak-ost'
+        --configKeyValues '' --early-forward-policy 'never' --fairmq-ipc-prefix '@'
+        --fairmq-rate-logging '0' --fairmq-recv-buffer-size '4' --fairmq-send-buffer-size
+        '4' --forwarding-destination 'drop' --forwarding-policy 'dangling' --host
+        '' --io-threads '1' --labels '' --local-batch '' --override-values '' --pipeline
+        '' --readers '1' --remote-batch '' --severity 'info' --shm-allocation 'rbtree_best_fit'
+        --shm-metadata-msg-size '0' --shm-mlock-segment 'false' --shm-mlock-segment-on-creation
+        'false' --shm-no-cleanup 'false' --shm-segment-id '0' --shm-zero-segment 'false'
+        --signposts '' --spawners '1' --stacktrace-on-signal 'simple' --timeframes-rate-limit
+        '0' --timeframes-rate-limit-ipcid -1 --workflow-suffix '' --qcConfiguration
+        '{}' --runNumber '' -S $CONTROL_OCCPLUGIN_ROOT/lib/ -P OCClite --color false
+        --control-port 31000
+      command:
+      - bash
+      - -c
+      env:
+      - name: O2_DETECTOR
+        value: TST
+      - name: O2_PARTITION
+        value: 34XmVqDQCsU
+      - name: HOME
+        value: /tmp
+      - name: O2_SYSTEM
+        value: FLP
+      - name: O2_ROLE
+        value: mtichak-ost
+      - name: OCC_CONTROL_PORT
+        value: "31000"
+      image: gitlab-registry.cern.ch/pkonopka/dockerfiles/flp-sw:20260729-1
+      imagePullPolicy: IfNotPresent
+      name: dpl
+      resources: {}
+      securityContext:
+        privileged: true
+        runAsGroup: 1100
+        runAsUser: 1100
+      volumeMounts:
+      - mountPath: /dev/shm
+        name: host-shm
+      - mountPath: /etc/group
+        name: group
+        readOnly: true
+      - mountPath: /etc/passwd
+        name: passwd
+        readOnly: true
+      - mountPath: /tmp
+        name: tmp
+      - mountPath: /lib/modules
+        name: modules
+    hostIPC: true
+    hostNetwork: true
+    imagePullSecrets:
+    - name: gitlab-registry-secret
+    nodeName: mtichak-ost.cern.ch
+    securityContext:
+      fsGroup: 1100
+      supplementalGroups:
+      - 10
+      - 1105
+    volumes:
+    - hostPath:
+        path: /dev/shm
+        type: Directory
+      name: host-shm
+    - hostPath:
+        path: /etc/group
+      name: group
+    - hostPath:
+        path: /etc/passwd
+      name: passwd
+    - hostPath:
+        path: /tmp
+      name: tmp
+    - hostPath:
+        path: /lib/modules
+      name: modules
+  state: standby

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,18 @@ 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}-(.+)$`)
+
 // k8sEnvRegistry maps ECS environment IDs to K8s custom Environment names.
 type k8sEnvRegistry struct {
 	mu   sync.RWMutex
@@ -83,6 +95,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 +145,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 +220,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 +254,49 @@ 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 {
+			// Right now ports are allocated only for jitted tasks. This way is used
+			// because ports are being assigned by Mesos. Right now we assume that readout
+			// and dd (stfbuilder/sender) have ports assigned in the TaskTemplate and
+			// thus they don't need portAllocator. This might change.
+			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. However
+			// it might be changed in the future.
+			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)
 	}

docs/kubernetes_ecs.md

@@ -51,6 +51,14 @@ To deploy a Kubernetes task you can in principle reuse existing control-workflow
 with reusing existing manifests, for example bad quotation of arguments: Mesos runs tasks via shell, which interpreted and stripped quotation from commands. For this reason you can find ready-to-use manifests with all necessary changes
 inside `control-operator/ecs-manifests/control-workflows/*kube-direct*`.
 
+### TaskTemplates
+
+Since some tasks are deployed with known values ahead of time (image, env vars, ...), we added the `TaskTemplate` CR, which stores this data in the cluster so the user doesn't have to define everything every time. You can find all the `TaskTemplate`s in the `ecs-manifests/task-templates` folder. Currently we have templates for DPL, Readout, and StfBuilder/Sender. These templates must be applied to the cluster for the ECS bridge to work.
+
+### DPL and JIT
+
+DPL tasks work similarly to those mentioned before: they use `kubernetes_fairmq` in their workflow YAML manifests. However, there is a problem: JIT tasks' manifests are created the moment they are run. So for now you need to first run the workflow to generate those tasks, then change the control mode after they've been generated. This will be changed in the future.
+
 ## Running tasks (`KubectlTask`)
 
 This method is obsoleted by direct ECS <-> Kubernetes bridge. However it should still work