diff --git a/control-operator/api/v1alpha1/environment_types.go b/control-operator/api/v1alpha1/environment_types.go index 2400adda..330fbde1 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 be25cf23..e3184eb4 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/kubernetes-manifests/dpl-qccheck-test.yaml b/control-operator/ecs-manifests/kubernetes-manifests/dpl-qccheck-test.yaml new file mode 100644 index 00000000..69a16845 --- /dev/null +++ b/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 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 00000000..2a5adeff --- /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 automatically by TaskController diff --git a/control-operator/ecs-manifests/task-templates/readout-tasktemplate.yaml b/control-operator/ecs-manifests/task-templates/readout-tasktemplate.yaml index 71c0dc91..16844741 100644 --- a/control-operator/ecs-manifests/task-templates/readout-tasktemplate.yaml +++ b/control-operator/ecs-manifests/task-templates/readout-tasktemplate.yaml @@ -62,4 +62,4 @@ spec: - name: gitlab-registry-secret control: mode: "direct" - # port: ${OCC_CONTROL_PORT} # to be filled in + # port: ${OCC_CONTROL_PORT} # to be filled in automatically by TaskController 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 cc7136c7..42e1b7a4 100644 --- a/control-operator/ecs-manifests/task-templates/stfbuilder-senderoutput-tasktemplate.yaml +++ b/control-operator/ecs-manifests/task-templates/stfbuilder-senderoutput-tasktemplate.yaml @@ -64,7 +64,7 @@ spec: - name: gitlab-registry-secret control: mode: "fairmq" - # port: ${OCC_CONTROL_PORT} #to be filled in + # port: ${OCC_CONTROL_PORT} # to be filled in automatically by TaskController bind: - name: buildertosender type: push diff --git a/control-operator/ecs-manifests/task-templates/stfsender-tasktemplate.yaml b/control-operator/ecs-manifests/task-templates/stfsender-tasktemplate.yaml index 87111c25..64d7f532 100644 --- a/control-operator/ecs-manifests/task-templates/stfsender-tasktemplate.yaml +++ b/control-operator/ecs-manifests/task-templates/stfsender-tasktemplate.yaml @@ -71,4 +71,4 @@ spec: - name: gitlab-registry-secret control: mode: "fairmq" - # port: ${OCC_CONTROL_PORT} # to be filled in + # port: ${OCC_CONTROL_PORT} # to be filled in automatically by TaskController diff --git a/control-operator/internal/controller/environment_controller.go b/control-operator/internal/controller/environment_controller.go index a605b37a..702b13d0 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 0f64cbac..b4a129df 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 6600d15d..c5a22bad 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,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>-" (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