From 16d497a551db3f3fd4214f2132ded194a4eb2112 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Michal=20Tich=C3=A1k?= Date: Tue, 18 Aug 2026 23:04:52 +0200 Subject: [PATCH 1/5] [task-controller] properly respect *_IMG for deployment via Makefile --- .../config/manager/environment-manager/kustomization.yaml | 5 ----- .../config/manager/task-manager/kustomization.yaml | 5 ----- 2 files changed, 10 deletions(-) diff --git a/control-operator/config/manager/environment-manager/kustomization.yaml b/control-operator/config/manager/environment-manager/kustomization.yaml index 31c476c0f..3349a34d4 100644 --- a/control-operator/config/manager/environment-manager/kustomization.yaml +++ b/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 diff --git a/control-operator/config/manager/task-manager/kustomization.yaml b/control-operator/config/manager/task-manager/kustomization.yaml index 51192d22e..578dd785e 100644 --- a/control-operator/config/manager/task-manager/kustomization.yaml +++ b/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 From 12232fcd18afa1531c6d78274f766d2ee252978e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Michal=20Tich=C3=A1k?= Date: Tue, 18 Aug 2026 23:05:21 +0200 Subject: [PATCH 2/5] [control-operator] minor fixes to TaskTemplates and test_env yamls --- control-operator/ecs-manifests/environment/test_env.yaml | 3 +-- .../ecs-manifests/task-templates/readout-tasktemplate.yaml | 1 - .../task-templates/stfbuilder-senderoutput-tasktemplate.yaml | 1 - .../ecs-manifests/task-templates/stfsender-tasktemplate.yaml | 1 - 4 files changed, 1 insertion(+), 5 deletions(-) diff --git a/control-operator/ecs-manifests/environment/test_env.yaml b/control-operator/ecs-manifests/environment/test_env.yaml index ae3c21eaf..8e0f94821 100644 --- a/control-operator/ecs-manifests/environment/test_env.yaml +++ b/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: diff --git a/control-operator/ecs-manifests/task-templates/readout-tasktemplate.yaml b/control-operator/ecs-manifests/task-templates/readout-tasktemplate.yaml index 71c0dc91b..2db71f56c 100644 --- a/control-operator/ecs-manifests/task-templates/readout-tasktemplate.yaml +++ b/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: diff --git a/control-operator/ecs-manifests/task-templates/stfbuilder-senderoutput-tasktemplate.yaml b/control-operator/ecs-manifests/task-templates/stfbuilder-senderoutput-tasktemplate.yaml index cc7136c7e..c097c4f6a 100644 --- a/control-operator/ecs-manifests/task-templates/stfbuilder-senderoutput-tasktemplate.yaml +++ b/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 diff --git a/control-operator/ecs-manifests/task-templates/stfsender-tasktemplate.yaml b/control-operator/ecs-manifests/task-templates/stfsender-tasktemplate.yaml index 87111c255..6efbab4d0 100644 --- a/control-operator/ecs-manifests/task-templates/stfsender-tasktemplate.yaml +++ b/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 From a1b11f69bd22b6dea72919b4272dad72e1710c55 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Michal=20Tich=C3=A1k?= Date: Mon, 17 Aug 2026 13:49:05 +0200 Subject: [PATCH 3/5] running dpl in k8s vis ECS --- .../api/v1alpha1/environment_types.go | 1 + .../bases/aliecs.alice.cern_environments.yaml | 2 + .../task-templates/dpl-tasktemplate.yaml | 62 +++++++++++ .../controller/environment_controller.go | 15 ++- core/config.go | 2 + core/task/managerk8s.go | 100 ++++++++++++++++-- 6 files changed, 170 insertions(+), 12 deletions(-) create mode 100644 control-operator/ecs-manifests/task-templates/dpl-tasktemplate.yaml diff --git a/control-operator/api/v1alpha1/environment_types.go b/control-operator/api/v1alpha1/environment_types.go index 2400addae..330fbde11 100644 --- a/control-operator/api/v1alpha1/environment_types.go +++ b/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"` diff --git a/control-operator/config/crd/bases/aliecs.alice.cern_environments.yaml b/control-operator/config/crd/bases/aliecs.alice.cern_environments.yaml index be25cf235..e3184eb46 100644 --- a/control-operator/config/crd/bases/aliecs.alice.cern_environments.yaml +++ b/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: diff --git a/control-operator/ecs-manifests/task-templates/dpl-tasktemplate.yaml b/control-operator/ecs-manifests/task-templates/dpl-tasktemplate.yaml new file mode 100644 index 000000000..1c48ee0a4 --- /dev/null +++ b/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 diff --git a/control-operator/internal/controller/environment_controller.go b/control-operator/internal/controller/environment_controller.go index a605b37ae..702b13d0c 100644 --- a/control-operator/internal/controller/environment_controller.go +++ b/control-operator/internal/controller/environment_controller.go @@ -57,9 +57,15 @@ 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 } @@ -67,6 +73,11 @@ func (r *EnvironmentReconciler) runTasksFromReferenceOnNode(ctx context.Context, 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 } diff --git a/core/config.go b/core/config.go index 0f64cbac9..b4a129dfa 100644 --- a/core/config.go +++ b/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) diff --git a/core/task/managerk8s.go b/core/task/managerk8s.go index 6600d15dc..8250fe347 100644 --- a/core/task/managerk8s.go +++ b/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>-" (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