diff --git a/control-operator/README.md b/control-operator/README.md index 91fc6b931..475060914 100644 --- a/control-operator/README.md +++ b/control-operator/README.md @@ -1,12 +1,12 @@ # operator -Folder with operators regarding Task and Environment deployment. +Folder with operators for Task and Environment deployment. For a better understanding of how the operators are used by ECS and the Control repository, see [docs/kubernetes_ecs.md](../docs/kubernetes_ecs.md). ## Description -In order to deploy Task and Environment workflows to the k8s cluster you need controllers and operators -controlling custom CRDs defining ALICE custom workload. This Folder defines and implements all moving parts together with Makefile -to build, deploy, install CRDs and operators. +In order to deploy Task and Environment workflows to the k8s cluster, you need controllers and operators +controlling the CRDs that define ALICE custom workloads. This folder defines and implements all the moving parts, together with a Makefile +to build, deploy, and install the CRDs and operators. ## Architecture @@ -18,26 +18,38 @@ The operator is split into two separate binaries with different deployment strat ## Getting Started -You’ll need a Kubernetes cluster to run against. You can use [KIND](https://sigs.k8s.io/kind) to get a local cluster for testing, or run against a remote cluster. Author had the most success with K3s [see](/docs/kubernetes_ecs.md). -**Note:** Your controller will automatically use the current context in your kubeconfig (usually ~/.kube/config) file (i.e. whatever cluster `kubectl cluster-info` shows). +You’ll need a Kubernetes cluster to run against. You can use [KIND](https://sigs.k8s.io/kind) to get a local cluster for testing, or run against a remote cluster. The author had the most success with K3s ([see](/docs/kubernetes_ecs.md)). +**Note:** Your controller will automatically use the current context in your kubeconfig file (usually `~/.kube/config`) (i.e. whatever cluster `kubectl cluster-info` shows). ### Running on the cluster -Following commands show basic use of Makefile. However this isn't exhaustive list. +#### Namespaces -1. Install Instances of Custom Resources: +This project expects two namespaces: `alice-tasks` and `operator-system`. `alice-tasks` is where the actual `Task` resources run, and it must be created before the first Task is deployed. `operator-system` is where all operator pods run, and it is created during deployment to the cluster. + +#### Secrets + +When deploying operator containers, you are required to set up proper secrets to access the registry where the operator images reside. At the moment they are built by a GitHub pipeline and pushed to the AliceO2Group GitHub [package registry](https://github.com/orgs/AliceO2Group/packages). + +The second required secret must be set up to access images used by `Task` CRs. At the moment these images are located in a GitLab registry, and the secret name is hardcoded in this repository as `gitlab-registry-secret`. + +#### make commands + +The following commands show basic usage of the Makefile. However, this isn't an exhaustive list. + +- Install instances of Custom Resources: ```sh kubectl apply -f config/samples/ ``` -1. Build and push your images. Default image tags are defined in the Makefile via `TASK_IMG` and `ENVIRONMENT_IMG`. Override them only if you want to use a different registry or tag: +- Build and push your images. Default image tags are defined in the Makefile via `TASK_IMG` and `ENVIRONMENT_IMG`. Override them only if you want to use a different registry or tag: ```sh make docker-build docker-push TASK_IMG=/task-manager:tag ENVIRONMENT_IMG=/environment-manager:tag ``` -1. Deploy the controllers to the cluster. Uses the same `TASK_IMG` and `ENVIRONMENT_IMG` defaults, override them if needed. Images are pulled using the secret named by `PULL_SECRET` (default `ghcr-pull`), which must exist in the `operator-system` namespace: +- Deploy the controllers to the cluster. Uses the same `TASK_IMG` and `ENVIRONMENT_IMG` defaults; override them if needed. Images are pulled using the secret named by `PULL_SECRET` (default `ghcr-pull`), which must exist in the `operator-system` namespace: ```sh make deploy TASK_IMG=/task-manager:tag ENVIRONMENT_IMG=/environment-manager:tag PULL_SECRET= @@ -45,7 +57,7 @@ make deploy TASK_IMG=/task-manager:tag ENVIRONMENT_IMG= make run-task ``` -**NOTE:** You can also install CRDs and run in one step: `make install run-environment` or `make install run-task` +**NOTE:** You can also install the CRDs and run in one step: `make install run-environment` or `make install run-task`. ### Modifying the API definitions @@ -100,9 +112,9 @@ If you are editing the API definitions, generate the manifests such as CRs or CR make manifests ``` -**NOTE:** Run `make --help` for more information on all potential `make` targets +**NOTE:** Run `make --help` for more information on all available `make` targets. -More information can be found via the [Kubebuilder Documentation](https://book.kubebuilder.io/introduction.html) +More information can be found in the [Kubebuilder Documentation](https://book.kubebuilder.io/introduction.html). ## License 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/control-workflows/stfbuilder-kube-direct.yaml b/control-operator/ecs-manifests/control-workflows/stfbuilder-kube-direct.yaml new file mode 100644 index 000000000..a50ac549b --- /dev/null +++ b/control-operator/ecs-manifests/control-workflows/stfbuilder-kube-direct.yaml @@ -0,0 +1,65 @@ +name: stfbuilder +control: + mode: kubernetes_fairmq +defaults: + detector: TST + rdh_version: "6" + user: flp + fmq_severity: info + dd_detector_subspec: feeid + dd_discovery_endpoint: "no-op://" # location of consul docker instance + log_task_stdout: none + log_task_stderr: none + stfb_dd_mode: "physics" + stfb_fee_mask: "0xffff" + stfb_enable_datasink: "false" + stfb_datasink_dir: "/tmp" + stfb_datasink_filesize: "2048" + monitoring_dd_interval: "5" + _module_cmdline: >- + source /etc/profile.d/modules.sh && MODULEPATH={{ modulepath }} module load DataDistribution Control-OCCPlugin && + StfBuilder + _plain_cmdline: "{{ o2_install_path }}/bin/StfBuilder" +wants: + cpu: 0.15 + memory: 128 +bind: + - name: dpl-chan + type: push + rateLogging: "{{ fmq_rate_logging }}" + transport: shmem + addressing: ipc + sndBufSize: "4" + global: "readout-proxy-{{ it }}" +command: + env: + - O2_INFOLOGGER_MODE={{infologger_mode}} + - O2_DETECTOR={{ detector }} + - DATADIST_FEE_MASK={{ stfb_fee_mask }} + - O2_PARTITION={{ environment_id }} + stdout: "{{ log_task_stdout }}" + stderr: "{{ log_task_stderr }}" + shell: true + user: "{{ user }}" + value: "{{ len(modulepath)>0 ? _module_cmdline : _plain_cmdline }}" + arguments: + - "--session=default" + - "--shm-segment-id=2" + - "--shm-segment-size=33554432" + - "--transport=shmem" + - "--monitoring-backend={{ monitoring_dd_url }}" + - "--monitoring-process-interval={{ monitoring_dd_interval }}" + - "--discovery-partition={{ environment_id }}" + - "--discovery-endpoint={{ dd_discovery_endpoint }}" + - "--detector-rdh={{ rdh_version }}" + - "--detector-subspec={{ dd_detector_subspec }}" + - "--severity={{ fmq_severity }}" + - "--severity-infologger={{ infologger_severity }}" + - "--dpl-channel-name=dpl-chan" + - "--shm-monitor=false" + - "--detector={{ detector }}" + - "--run-type={{ stfb_dd_mode }}" + - "{{ stfb_enable_datasink == 'true' ? '--data-sink-enable' : ' ' }}" + - "{{ stfb_enable_datasink == 'true' ? '--data-sink-dir=' + stfb_datasink_dir : ' ' }}" + - "{{ stfb_enable_datasink == 'true' ? '--data-sink-max-stfs-per-file=0' : ' ' }}" + - "{{ stfb_enable_datasink == 'true' ? '--data-sink-max-file-size=' + stfb_datasink_filesize : ' ' }}" diff --git a/control-operator/ecs-manifests/kubernetes-ecs.md b/control-operator/ecs-manifests/kubernetes-ecs.md deleted file mode 100644 index e5608750d..000000000 --- a/control-operator/ecs-manifests/kubernetes-ecs.md +++ /dev/null @@ -1,35 +0,0 @@ -# ECS in Kubernetes - -For now we can run OCC tasks in Kubernetes (namely `readout`, `stfsender`, `stfbuilder-senderoutput`) using -the task controller inside the `control-operator` folder and `kubectltask` created as a version of `controllabletask` -for the Mesos executor wrapping `kubectl` tool. However as the `kubectl` requires manifests we used `control-operator/ecs-manifests` -to store these test manifests. There are more test files apart from these manifests. - -There are 3 subfolders inside `ecs-manifests`, namely `control-workflows`, `kubernetes-manifests` and `occ-configure-arguments`. - -### `control-workflows` - -In order to run given task from ECS you need to provide yaml template normally contained inside the `ControlWorkflows` -repository that is processed by ECS core and sent to Mesos framework that runs the given task on a given agent. -Inside the folder you can find files with suffixes `docker`, `kube`, `orig` appended to the name of the task they -are representing. These files are to be put into the `ControlWorkflows/tasks` so ECS can find those and run task -in proper way. Eg. if one is to run readout in Kubernetes copy (or symlink) `readout-kube.yaml` into the `ControlWorkflows` -directory under the name `readout.yaml` (same for the other tasks) - -### `kubernetes-manifests` - -Kubectltask requires Kubernetes manifests to pass to `kubectl`, these manifests are located in directory `kubernetes-manifests`. -There are two types of manifests `task.yaml` and `task-test.yaml`. The first one is the actual manifest to be used by -executors and kubectltask where environment variables substitution is used in a form of `${VAR}`. `task-test.yaml` -has the same form as templated manifest, but with actual test values to test container/binary inside the kubernetes. -Apply the manifest by invoking: - -```bash -kubectl apply -f task-test.yaml -``` - -### `occ-configure-arguments` - -yaml files containing data used by `peanut` to properly transition `readout`, `stfsender` and `stfbuilder-senderoutput` -to `CONFIGURED`. Use these files by either loading them into `peanut` with LoadConfiguration in TUI mode or -by using `--config` flag in CLI mode. 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 000000000..69a16845d --- /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 000000000..a45a82c17 --- /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/aliceo2group/dockerfiles/alma9-flp-node/o2suite:20261001-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 2db71f56c..5aa480497 100644 --- a/control-operator/ecs-manifests/task-templates/readout-tasktemplate.yaml +++ b/control-operator/ecs-manifests/task-templates/readout-tasktemplate.yaml @@ -17,7 +17,7 @@ spec: supplementalGroups: [10, 1105] containers: - name: readout - image: gitlab-registry.cern.ch/aliceo2group/dockerfiles/alma9-flp-node/readout:1 + image: gitlab-registry.cern.ch/aliceo2group/dockerfiles/alma9-flp-node/o2suite:20261001-1 command: ["/opt/o2/bin/o2-readout-exe"] env: - name: OCC_CONTROL_PORT @@ -61,4 +61,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 c097c4f6a..978389bad 100644 --- a/control-operator/ecs-manifests/task-templates/stfbuilder-senderoutput-tasktemplate.yaml +++ b/control-operator/ecs-manifests/task-templates/stfbuilder-senderoutput-tasktemplate.yaml @@ -19,7 +19,7 @@ spec: supplementalGroups: [10, 1105] containers: - name: stfbuilder - image: gitlab-registry.cern.ch/aliceo2group/dockerfiles/alma9-flp-node/dd:1 + image: gitlab-registry.cern.ch/aliceo2group/dockerfiles/alma9-flp-node/o2suite:20261001-1 command: [/opt/o2/bin/StfBuilder] env: - name: OCC_CONTROL_PORT @@ -63,7 +63,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/stfbuilder-tasktemplate.yaml b/control-operator/ecs-manifests/task-templates/stfbuilder-tasktemplate.yaml new file mode 100644 index 000000000..42d297624 --- /dev/null +++ b/control-operator/ecs-manifests/task-templates/stfbuilder-tasktemplate.yaml @@ -0,0 +1,77 @@ +apiVersion: aliecs.alice.cern/v1alpha1 +kind: TaskTemplate +metadata: + name: stfbuilder + namespace: alice-tasks +spec: + envVars: + - O2_DETECTOR + - O2_PARTITION + - O2_SYSTEM + - O2_ROLE + - O2_INFOLOGGER_MODE + - DATADIST_FEE_MASK + pod: + hostNetwork: true + hostIPC: true + securityContext: + fsGroup: 1100 + supplementalGroups: [10, 1105] + containers: + - name: stfbuilder + image: gitlab-registry.cern.ch/aliceo2group/dockerfiles/alma9-flp-node/o2suite:20261001-1 + command: [/opt/o2/bin/StfBuilder] + env: + - name: OCC_CONTROL_PORT + value: "47102" + 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 + bind: + - name: dpl-chan + type: push + rateLogging: "0" + addressing: ipc + transport: shmem + sndBufSize: 4 + global: "" + target: "" + connect: [] + properties: {} diff --git a/control-operator/ecs-manifests/task-templates/stfsender-tasktemplate.yaml b/control-operator/ecs-manifests/task-templates/stfsender-tasktemplate.yaml index 6efbab4d0..62efc2c7c 100644 --- a/control-operator/ecs-manifests/task-templates/stfsender-tasktemplate.yaml +++ b/control-operator/ecs-manifests/task-templates/stfsender-tasktemplate.yaml @@ -26,7 +26,7 @@ spec: supplementalGroups: [10, 1105] containers: - name: stfsender - image: gitlab-registry.cern.ch/aliceo2group/dockerfiles/alma9-flp-node/dd:1 + image: gitlab-registry.cern.ch/aliceo2group/dockerfiles/alma9-flp-node/o2suite:20261001-1 command: ["numactl"] env: - name: OCC_CONTROL_PORT @@ -70,4 +70,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 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/control-operator/pkg/client/client.go b/control-operator/pkg/client/client.go index 41235f48d..d81b9d77b 100644 --- a/control-operator/pkg/client/client.go +++ b/control-operator/pkg/client/client.go @@ -27,6 +27,7 @@ package client import ( "context" "fmt" + "maps" "github.com/AliceO2Group/Control/control-operator/api/v1alpha1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -93,6 +94,13 @@ func (c *Client) UpdateTask(ctx context.Context, task *v1alpha1.Task) error { return c.client.Update(ctx, task) } +// SetTaskArguments replaces the Task's spec.arguments with the given transition arguments +func (c *Client) SetTaskArguments(ctx context.Context, task *v1alpha1.Task, args map[string]string) error { + patch := crClient.MergeFrom(task.DeepCopy()) + task.Spec.Arguments = maps.Clone(args) + return c.client.Patch(ctx, task, patch) +} + func (c *Client) DeleteTask(ctx context.Context, name string) error { return c.client.Delete(ctx, &v1alpha1.Task{ ObjectMeta: metav1.ObjectMeta{Name: name, Namespace: c.namespace}, diff --git a/core/config.go b/core/config.go index 7c116d312..6e14c97af 100644 --- a/core/config.go +++ b/core/config.go @@ -132,6 +132,7 @@ func setDefaults() error { viper.SetDefault("enableKubernetes", false) viper.SetDefault("kubeNamespace", "alice-tasks") viper.SetDefault("kubeConfig", "/home/aliecs/.kube/config") + viper.SetDefault("jitK8sBasePort", uint16(32000)) return nil } @@ -206,6 +207,7 @@ func setFlags() error { pflag.Bool("enableKubernetes", viper.GetBool("enableKubernetes"), "Turn on Kubernetes support (requires valid kubeNamespace and kubeConfig)") pflag.String("kubeNamespace", viper.GetString("kubeNamespace"), "Kubernetes namespace in which ECS creates Task and Environment objects") pflag.String("kubeConfig", viper.GetString("kubeConfig"), "Path to the kubeconfig used to access the Kubernetes cluster (empty uses in-cluster config)") + 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/manager.go b/core/task/manager.go index 706a62863..8c83bccca 100644 --- a/core/task/manager.go +++ b/core/task/manager.go @@ -905,18 +905,20 @@ func (m *Manager) configureTasks(envId uid.ID, tasks Tasks) error { } func (m *Manager) transitionTasks(envId uid.ID, tasks Tasks, src string, event string, dest string, commonArgs controlcommands.PropertyMap) error { - var mesosTasks Tasks + var mesosTasks, k8sTasks Tasks for _, t := range tasks { - if t.GetControlMode() != controlmode.KUBERNETES_DIRECT && - t.GetControlMode() != controlmode.KUBERNETES_FAIRMQ { + if t.GetControlMode() == controlmode.KUBERNETES_DIRECT || + t.GetControlMode() == controlmode.KUBERNETES_FAIRMQ { + k8sTasks = append(k8sTasks, t) + } else { mesosTasks = append(mesosTasks, t) } } if len(tasks) != len(mesosTasks) { log.WithField("partition", envId).Infof("Transitioning k8s environment from %s, to %s, event: %s", src, dest, event) - if err := m.transitionAndWaitK8sEnvState(context.Background(), envId, strings.ToLower(dest)); err != nil { + if err := m.pushArgsAndTransitionK8sEnvState(context.Background(), envId, k8sTasks, strings.ToLower(dest), commonArgs); err != nil { return err } } diff --git a/core/task/managerk8s.go b/core/task/managerk8s.go index 9a6039b2f..fb21ba4d1 100644 --- a/core/task/managerk8s.go +++ b/core/task/managerk8s.go @@ -28,6 +28,8 @@ import ( "context" "fmt" "maps" + "regexp" + "strconv" "strings" "sync" "time" @@ -41,9 +43,11 @@ import ( "github.com/AliceO2Group/Control/common/utils/uid" "github.com/AliceO2Group/Control/control-operator/api/v1alpha1" k8sclient "github.com/AliceO2Group/Control/control-operator/pkg/client" + "github.com/AliceO2Group/Control/core/controlcommands" "github.com/AliceO2Group/Control/core/task/channel" "github.com/AliceO2Group/Control/core/task/sm" "github.com/AliceO2Group/Control/core/task/taskclass" + "github.com/k0kubun/pp" "github.com/spf13/viper" ) @@ -52,8 +56,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 +97,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. func newK8sClientFromViper() (*k8sclient.Client, error) { namespace := viper.GetString("kubeNamespace") @@ -111,7 +146,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 @@ -186,8 +221,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 @@ -216,18 +255,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