Run dpl workflows from ECS in k8s

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/cmd/task-manager/main.go

@@ -57,11 +57,14 @@ func main() {
 	var metricsAddr string
 	var enableLeaderElection bool
 	var probeAddr string
+	var maxConcurrentReconciles int
 	flag.StringVar(&metricsAddr, "metrics-bind-address", ":9082", "The address the metric endpoint binds to.")
 	flag.StringVar(&probeAddr, "health-probe-bind-address", ":9083", "The address the probe endpoint binds to.")
 	flag.BoolVar(&enableLeaderElection, "leader-elect", false,
 		"Enable leader election for controller manager. "+
 			"Enabling this will ensure there is only one active controller manager.")
+	flag.IntVar(&maxConcurrentReconciles, "max-concurrent-reconciles", 1,
+		"The maximum number of concurrent Reconciles which can be run for the Task controller.")
 	opts := zap.Options{
 		Development: true,
 	}
@@ -90,10 +93,11 @@ func main() {
 	}
 
 	if err = (&controller.TaskReconciler{
-		Client:   mgr.GetClient(),
-		Scheme:   mgr.GetScheme(),
-		Recorder: mgr.GetEventRecorderFor("task-controller"),
-		NodeName: nodeName,
+		Client:                  mgr.GetClient(),
+		Scheme:                  mgr.GetScheme(),
+		Recorder:                mgr.GetEventRecorderFor("task-controller"),
+		NodeName:                nodeName,
+		MaxConcurrentReconciles: maxConcurrentReconciles,
 	}).SetupWithManager(mgr); err != nil {
 		setupLog.Error(err, "unable to create controller", "controller", "Task")
 		os.Exit(1)

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/config/manager/environment-manager/kustomization.yaml

@@ -3,8 +3,3 @@ kind: Kustomization
 
 resources:
 - environment-manager.yaml
-
-images:
-- name: environment-manager
-  newName: gitlab-registry.cern.ch/aliceo2group/dockerfiles/aliecs/environment-manager
-  newTag: latest

control-operator/config/manager/task-manager/kustomization.yaml

@@ -3,8 +3,3 @@ kind: Kustomization
 
 resources:
 - task-manager.yaml
-
-images:
-- name: task-manager
-  newName: gitlab-registry.cern.ch/aliceo2group/dockerfiles/aliecs/task-manager
-  newTag: latest

control-operator/ecs-manifests/environment/test_env.yaml

@@ -6,8 +6,7 @@ metadata:
 
 taskTemplates:
   tasks:
-    mtichak-ost.cern.ch:
-    # mtichak-ost:
+    mtichak-ost:
       # mtichak-xps-cern:
       # - name: readout
       #   argsCLI:

control-operator/ecs-manifests/kubernetes-manifests/dpl-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/ecs-manifests/task-templates/readout-tasktemplate.yaml

@@ -7,7 +7,6 @@ spec:
   envVars:
     - O2_DETECTOR
     - O2_PARTITION
-    - OCC_CONTROL_PORT
     - O2_SYSTEM
     - O2_ROLE
   pod:

control-operator/ecs-manifests/task-templates/stfbuilder-senderoutput-tasktemplate.yaml

@@ -7,7 +7,6 @@ spec:
   envVars:
     - O2_DETECTOR
     - O2_PARTITION
-    - OCC_CONTROL_PORT
     - O2_SYSTEM
     - O2_ROLE
     - O2_INFOLOGGER_MODE

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

@@ -7,7 +7,6 @@ spec:
   envVars:
     - O2_DETECTOR
     - O2_PARTITION
-    - OCC_CONTROL_PORT
     - O2_SYSTEM
     - O2_ROLE
     - O2_INFOLOGGER_MODE

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
 		}