Skip to content

[OCTRL-1091] Environment controller changes for ECS #828

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
justonedev1 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[OCTRL-1091] Environment controller changes for ECS #828

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions control-operator/api/v1alpha1/environment_types.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,7 @@ type EnvironmentStatus struct {

type TaskReference struct {
Name string `json:"name"`
TaskID string `json:"taskID,omitempty"`
Env []v1.EnvVar `json:"env"`
ArgsCLI []string `json:"argsCLI"`
ArgsTransition map[string]string `json:"argsTransition"`
Expand Down
2 changes: 2 additions & 0 deletions control-operator/config/crd/bases/aliecs.alice.cern_environments.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4100,6 +4100,8 @@ spec:
type: array
name:
type: string
taskID:
type: string
required:
- argsCLI
- argsTransition
Expand Down
7 changes: 5 additions & 2 deletions control-operator/ecs-manifests/environment/test_env.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,10 +2,12 @@ apiVersion: aliecs.alice.cern/v1alpha1
kind: Environment
metadata:
name: 31jdwll4xl6
namespace: alice-tasks

taskTemplates:
tasks:
mtichak-ost.cern.ch:
# mtichak-ost.cern.ch:
mtichak-ost:
# mtichak-xps-cern:
# - name: readout
# argsCLI:
Expand Down Expand Up @@ -187,7 +189,8 @@ taskTemplates:
spec:
state: standby
tasks:
mtichak-ost.cern.ch:
# mtichak-ost.cern.ch:
mtichak-ost:
- name: readout
spec:
pod:
Expand Down
5 changes: 4 additions & 1 deletion control-operator/ecs-manifests/kubernetes-manifests/readout-test.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,9 @@ apiVersion: aliecs.alice.cern/v1alpha1
kind: Task
metadata:
name: readout
namespace: alice-tasks
spec:
nodeName: mtichak-ost.cern.ch

@knopers8 knopers8 Jul 8, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am a bit lost why ".cern.ch" is needed here, but should be absent in the environment templates.

arguments:
chans.readout.0.address: ipc://@o2ipc-d7ef3du8ndmd7n8j7l2g
chans.readout.0.autoBind: "0"
Expand All @@ -25,7 +27,7 @@ spec:
supplementalGroups: [10, 1105]
containers:
- name: readout
image: gitlab-registry.cern.ch/aliceo2group/dockerfiles/alma9-flp-node:2
image: gitlab-registry.cern.ch/aliceo2group/dockerfiles/alma9-flp-node/readout:1
command: ["/opt/o2/bin/o2-readout-exe"]
args: ["/etc/readout/readout-cfg.ini"]
securityContext:
Expand Down Expand Up @@ -99,6 +101,7 @@ apiVersion: v1
kind: ConfigMap
metadata:
name: readout-config
namespace: alice-tasks
data:
readout-cfg.ini: |
# Example configuration file for o2-readout-exe
Expand Down
2 changes: 2 additions & 0 deletions control-operator/ecs-manifests/kubernetes-manifests/stfbuilder-senderoutput-test.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,9 @@ apiVersion: aliecs.alice.cern/v1alpha1
kind: Task
metadata:
name: stfbuilder-senderoutput
namespace: alice-tasks
spec:
nodeName: mtichak-ost.cern.ch
arguments:
chans.buildertosender.0.address: ipc://@o2ipc-d7ef3du8ndmd7n8j7l30
chans.buildertosender.0.autoBind: "0"
Expand Down
2 changes: 2 additions & 0 deletions control-operator/ecs-manifests/kubernetes-manifests/stfsender-test.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,9 @@ apiVersion: aliecs.alice.cern/v1alpha1
kind: Task
metadata:
name: stfsender
namespace: alice-tasks
spec:
nodeName: mtichak-ost.cern.ch
arguments:
chans.buildertosender.0.address: ipc://@o2ipc-d7ef3du8ndmd7n8j7l30
chans.buildertosender.0.method: connect
Expand Down
11 changes: 9 additions & 2 deletions control-operator/ecs-manifests/task-templates/readout-tasktemplate.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,22 +2,30 @@ apiVersion: aliecs.alice.cern/v1alpha1
kind: TaskTemplate
metadata:
name: readout
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]
Comment on lines +15 to +17

@knopers8 knopers8 Jul 8, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it looks like this does not correspond to any of the three changes?

- matching node names in cluster from partial names yaml templates
- add taskID annotation to be able to filter tasks
- OCC_CONTROL_PORT overwrites Spec.Control.Port of task

can you extract to a separate PR or amend the commit?

containers:
- name: readout
image: gitlab-registry.cern.ch/aliceo2group/dockerfiles/alma9-flp-node/readout:1
command: ["/opt/o2/bin/o2-readout-exe"]
env:
- name: OCC_CONTROL_PORT
value: "47101"
securityContext:
privileged: true
runAsUser: 1100
runAsGroup: 1100
Comment on lines +27 to +28

@knopers8 knopers8 Jul 8, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

these neither

volumeMounts:
- name: host-shm
mountPath: /dev/shm
Expand Down Expand Up @@ -53,4 +61,3 @@ spec:
- name: gitlab-registry-secret
control:
mode: "direct"
# port: ${OCC_CONTROL_PORT} # to be filled in
9 changes: 8 additions & 1 deletion control-operator/ecs-manifests/task-templates/stfbuilder-senderoutput-tasktemplate.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,11 @@ apiVersion: aliecs.alice.cern/v1alpha1
kind: TaskTemplate
metadata:
name: stfbuilder-senderoutput
namespace: alice-tasks
spec:
envVars:
- O2_DETECTOR
- O2_PARTITION
- OCC_CONTROL_PORT
- O2_SYSTEM
- O2_ROLE
- O2_INFOLOGGER_MODE
Expand All @@ -15,12 +15,19 @@ spec:
hostNetwork: true
hostIPC: true
securityContext:
fsGroup: 1100
supplementalGroups: [10, 1105]
containers:
- name: stfbuilder
image: gitlab-registry.cern.ch/aliceo2group/dockerfiles/alma9-flp-node/dd: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
Expand Down
9 changes: 8 additions & 1 deletion control-operator/ecs-manifests/task-templates/stfsender-tasktemplate.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,11 +2,11 @@ apiVersion: aliecs.alice.cern/v1alpha1
kind: TaskTemplate
metadata:
name: stfsender
namespace: alice-tasks
spec:
envVars:
- O2_DETECTOR
- O2_PARTITION
- OCC_CONTROL_PORT
- O2_SYSTEM
- O2_ROLE
- O2_INFOLOGGER_MODE
Expand All @@ -22,12 +22,19 @@ spec:
hostNetwork: true
hostIPC: true
securityContext:
fsGroup: 1100
supplementalGroups: [10, 1105]
containers:
- name: stfsender
image: gitlab-registry.cern.ch/aliceo2group/dockerfiles/alma9-flp-node/dd:1
command: ["numactl"]
env:
- name: OCC_CONTROL_PORT
value: "47103"
securityContext:
privileged: true
runAsUser: 1100
runAsGroup: 1100
volumeMounts:
- name: host-shm
mountPath: /dev/shm
Expand Down
93 changes: 64 additions & 29 deletions control-operator/internal/controller/environment_controller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import (
"maps"
"slices"
"strconv"
"strings"

v1 "k8s.io/api/core/v1"
k8serrors "k8s.io/apimachinery/pkg/api/errors"
Expand All @@ -53,7 +54,7 @@ type EnvironmentReconciler struct {
}

func (r *EnvironmentReconciler) runTasksFromReferenceOnNode(ctx context.Context, taskReferences []aliecsv1alpha1.TaskReference,
nodename string, req ctrl.Request, environment *aliecsv1alpha1.Environment, log logr.Logger,
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)
Expand Down Expand Up @@ -85,31 +86,44 @@ func (r *EnvironmentReconciler) runTasksFromReferenceOnNode(ctx context.Context,

// TODO: regarding error handling. Is it correct to stop while handling one task? this will fail the whole deployment,
// which might not be desirable outcome especially for non-critical tasks
if foundIdx := slices.IndexFunc(taskReference.Env, func(envVar v1.EnvVar) bool { return envVar.Name == "OCC_CONTROL_PORT" }); foundIdx == -1 {
log.Error(fmt.Errorf("didn't find OCC_CONTROL_PORT in env"), "failed to fill in env vars from template")
return &reconcile.Result{}, nil
} else {
port, err := strconv.Atoi(taskReference.Env[foundIdx].Value)
if err != nil {
log.Error(fmt.Errorf("found OCC_CONTROL_PORT isn't convertible to number "), "failed to fill in env vars from template")
return &reconcile.Result{}, nil
}
task.Spec.Control.Port = port
}
task.Spec.Control.Port = template.Spec.Control.Port

if task.Spec.Arguments == nil {
task.Spec.Arguments = make(map[string]string)
}

// TODO: check for containers!

@knopers8 knopers8 Jul 8, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

remove todo?

task.Spec.Pod.Containers[0].Env = append(task.Spec.Pod.Containers[0].Env, taskReference.Env...)
existing := make(map[string]int, len(task.Spec.Pod.Containers[0].Env))
for i, e := range task.Spec.Pod.Containers[0].Env {
existing[e.Name] = i
}
for _, e := range taskReference.Env {
if i, ok := existing[e.Name]; ok {
task.Spec.Pod.Containers[0].Env[i] = e
} else {
task.Spec.Pod.Containers[0].Env = append(task.Spec.Pod.Containers[0].Env, e)
}
}
task.Spec.Pod.Containers[0].Args = append(task.Spec.Pod.Containers[0].Args, taskReference.ArgsCLI...)

for _, envVar := range task.Spec.Pod.Containers[0].Env {
if envVar.Name == "OCC_CONTROL_PORT" {
if port, err := strconv.Atoi(envVar.Value); err == nil {
task.Spec.Control.Port = port
}
break
}
}
maps.Copy(task.Spec.Arguments, taskReference.ArgsTransition)

task.Spec.Pod.NodeName = nodename
task.Spec.Pod.NodeName = resolvedNodename
task.Spec.NodeName = resolvedNodename
task.Spec.State = environment.Spec.State

task.Labels = labels(environment, nodename)
if taskReference.TaskID != "" {
task.Labels["taskID"] = taskReference.TaskID
}

if err := controllerutil.SetControllerReference(environment, task, r.Scheme); err != nil {
log.Error(err, "failed to set controller reference", "task", task.Name)
Expand All @@ -128,7 +142,7 @@ func (r *EnvironmentReconciler) runTasksFromReferenceOnNode(ctx context.Context,
}

func (r *EnvironmentReconciler) runTaskFromDefinitionOnNode(ctx context.Context, taskDefs []aliecsv1alpha1.TaskDefinition,
nodename string, req ctrl.Request, environment *aliecsv1alpha1.Environment, log logr.Logger,
nodename string, resolvedNodename string, req ctrl.Request, environment *aliecsv1alpha1.Environment, log logr.Logger,
) (*ctrl.Result, error) {
for _, taskDef := range taskDefs {
task := &aliecsv1alpha1.Task{}
Expand All @@ -148,7 +162,8 @@ func (r *EnvironmentReconciler) runTaskFromDefinitionOnNode(ctx context.Context,
maps.Copy(task.Spec.Arguments, taskDef.Spec.Arguments)
}

task.Spec.Pod.NodeName = nodename
task.Spec.Pod.NodeName = resolvedNodename
task.Spec.NodeName = resolvedNodename
task.Spec.State = environment.Spec.State

task.Labels = labels(environment, nodename)
Expand Down Expand Up @@ -203,24 +218,31 @@ func (r *EnvironmentReconciler) Reconcile(ctx context.Context, req ctrl.Request)
}

if environment.Status.State == "" {
nodeList := &v1.NodeList{}
if err := r.List(ctx, nodeList); err != nil {
return ctrl.Result{}, fmt.Errorf("listing nodes: %w", err)
}

for nodename, tasksReferences := range environment.TaskTemplates.Tasks {
log.Info("creating tasks for hostname from references", "hostname", nodename, "number of tasks", len(tasksReferences))
if res, err := r.checkNodeExistence(ctx, nodename); err != nil {
return res, err
resolvedName, err := resolveNodeName(nodeList.Items, nodename)
if err != nil {
return ctrl.Result{}, err
}

if res, err := r.runTasksFromReferenceOnNode(ctx, tasksReferences, nodename, req, environment, log); res != nil {
if res, err := r.runTasksFromReferenceOnNode(ctx, tasksReferences, nodename, resolvedName, req, environment, log); res != nil {
return *res, err
}
}

for nodename, taskTemplates := range environment.Spec.Tasks {
log.Info("creating tasks for hostname from definitions", "hostname", nodename, "number of tasks", len(taskTemplates))
if res, err := r.checkNodeExistence(ctx, nodename); err != nil {
return res, err
resolvedName, err := resolveNodeName(nodeList.Items, nodename)
if err != nil {
return ctrl.Result{}, err
}

if res, err := r.runTaskFromDefinitionOnNode(ctx, taskTemplates, nodename, req, environment, log); res != nil {
if res, err := r.runTaskFromDefinitionOnNode(ctx, taskTemplates, nodename, resolvedName, req, environment, log); res != nil {
return *res, err
}
}
Expand Down Expand Up @@ -268,15 +290,28 @@ func (r *EnvironmentReconciler) Reconcile(ctx context.Context, req ctrl.Request)
return ctrl.Result{}, nil
}

func (r *EnvironmentReconciler) checkNodeExistence(ctx context.Context, nodename string) (ctrl.Result, error) {
node := &v1.Node{}
if err := r.Get(ctx, types.NamespacedName{Name: nodename}, node); err != nil {
if k8serrors.IsNotFound(err) {
return ctrl.Result{}, fmt.Errorf("node %s not found in cluster", nodename)
// resolveNodeName resolves a short node name (e.g. "flp001") to the full name known to
// the cluster (e.g. "flp001.cern.ch") by checking whether any dot-separated component
// of a node's name matches exactly. Returns an error if zero or more than one node matches.
func resolveNodeName(nodes []v1.Node, nodename string) (string, error) {

@knopers8 knopers8 Jul 8, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please use "nodeName" or "nodename" consistently, i.e. "name" either always starting upper or lower case. IMO "nodeName" is better, since it's two words, but I leave the final decision to you.

var matches []string
for _, node := range nodes {
for _, part := range strings.Split(node.Name, ".") {
if part == nodename {
matches = append(matches, node.Name)
break
}
}
return ctrl.Result{}, err
}
return ctrl.Result{}, nil

switch len(matches) {
case 0:
return "", fmt.Errorf("node %q not found in cluster", nodename)
case 1:
return matches[0], nil
default:
return "", fmt.Errorf("node name %q is ambiguous, matched: %v", nodename, matches)

@knopers8 knopers8 Jul 8, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i appreciate being paranoic!

}
}

func aggregateState(tasks []aliecsv1alpha1.Task, previousState string) string {
Expand Down
Loading