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/cmd/task-manager/main.go b/control-operator/cmd/task-manager/main.go index 046a0b6f9..b3c45197b 100644 --- a/control-operator/cmd/task-manager/main.go +++ b/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) 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/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 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/kubernetes-manifests/dpl-test.yaml b/control-operator/ecs-manifests/kubernetes-manifests/dpl-test.yaml new file mode 100644 index 000000000..69a16845d --- /dev/null +++ b/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 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/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 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/internal/controller/task_controller.go b/control-operator/internal/controller/task_controller.go index a9a486a4f..ed04e8260 100644 --- a/control-operator/internal/controller/task_controller.go +++ b/control-operator/internal/controller/task_controller.go @@ -30,6 +30,7 @@ import ( "fmt" "reflect" "strings" + "sync" "time" v1 "k8s.io/api/core/v1" @@ -54,12 +55,14 @@ import ( // TaskReconciler reconciles a Task object type TaskReconciler struct { client.Client - Scheme *runtime.Scheme - Recorder record.EventRecorder - NodeName string + Scheme *runtime.Scheme + Recorder record.EventRecorder + NodeName string + MaxConcurrentReconciles int } -var clientsForContainers map[string]*OccClient = make(map[string]*OccClient) +// var clientsForContainers map[string]*OccClient = make(map[string]*OccClient) +var clientsForContainers sync.Map const taskFinalizer string = "aliecs.alice.cern/finalizer" @@ -136,7 +139,8 @@ func (r *TaskReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl. return ctrl.Result{}, nil } - if _, exists := clientsForContainers[t.Name]; !exists { + // if _, exists := clientsForContainers[t.Name]; !exists { + if _, exists := clientsForContainers.Load(t.Name); !exists { if existingPod.Status.PodIP == "" { log.Info("pod doesn't have IP yet, we wait for different event") return ctrl.Result{}, nil @@ -159,12 +163,13 @@ func (r *TaskReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl. // on them being implemented if t.Status.State == "" { log.V(1).Info("Status.State is empty, querying container") - client, exists := clientsForContainers[t.Name] + // client, exists := clientsForContainers[t.Name] + client, exists := clientsForContainers.Load(t.Name) if !exists { return ctrl.Result{Requeue: true}, nil } - stateReply, err := client.GetState(ctx) + stateReply, err := client.(*OccClient).GetState(ctx) if err != nil { log.Error(err, "Failed to GetState") return ctrl.Result{RequeueAfter: 5 * time.Second}, nil @@ -188,16 +193,18 @@ func (r *TaskReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl. // Handle Spec -> gRPC State Sync if t.Status.State != t.Spec.State { - client, exists := clientsForContainers[t.Name] + // client, exists := clientsForContainers[t.Name] + client, exists := clientsForContainers.Load(t.Name) if !exists { return ctrl.Result{Requeue: true}, nil } - stateReply, err := client.GetState(ctx) + stateReply, err := client.(*OccClient).GetState(ctx) if err != nil { log.Info("Failed to get state for sync, retrying in 5s", "error", err.Error()) - client.Close() - delete(clientsForContainers, t.Name) + client.(*OccClient).Close() + // delete(clientsForContainers, t.Name) + clientsForContainers.Delete(t.Name) return ctrl.Result{RequeueAfter: 5 * time.Second}, nil } @@ -208,9 +215,9 @@ func (r *TaskReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl. ) if t.Spec.Control.Mode == "fairmq" { - newState, transErr = client.FairMQTransitionRequest(ctx, stateReply.GetState(), t.Spec.State, t.Spec.Arguments) + newState, transErr = client.(*OccClient).FairMQTransitionRequest(ctx, stateReply.GetState(), t.Spec.State, t.Spec.Arguments) } else { - reply, err := client.TransitionRequest(ctx, stateReply.GetState(), t.Spec.State, t.Spec.Arguments) + reply, err := client.(*OccClient).TransitionRequest(ctx, stateReply.GetState(), t.Spec.State, t.Spec.Arguments) transErr = err if err == nil && reply.GetOk() { newState = strings.ToLower(reply.GetState()) @@ -268,7 +275,8 @@ func (r *TaskReconciler) createGRPCConsumer(ctx context.Context, t *aliecsv1alph return ctrl.Result{RequeueAfter: 5 * time.Second}, nil } - clientsForContainers[t.Name] = client + // clientsForContainers[t.Name] = client + clientsForContainers.Store(t.Name, client) if err := r.recordCondition(ctx, t, aliecsv1alpha1.ConditionGRPCConnected, metav1.ConditionTrue, "Connected", fmt.Sprintf("gRPC connection established to %s", addr)); err != nil { return ctrl.Result{}, err @@ -277,14 +285,15 @@ func (r *TaskReconciler) createGRPCConsumer(ctx context.Context, t *aliecsv1alph } func (r *TaskReconciler) consumeGRPCConsumerIfReady(ctx context.Context, t *aliecsv1alpha1.Task, log logr.Logger) ctrl.Result { - client, exists := clientsForContainers[t.Name] + // client, exists := clientsForContainers[t.Name] + client, exists := clientsForContainers.Load(t.Name) if !exists { log.Info("didn't found existing client, retrying ", "task", t.Name) return ctrl.Result{RequeueAfter: time.Second} } - if !client.ConsumeIfReady(ctx) { + if !client.(*OccClient).ConsumeIfReady(ctx) { log.Info("gRPC client is not ready, retrying in 5 seconds", "name", t.Name) return ctrl.Result{RequeueAfter: 5 * time.Second} } @@ -342,12 +351,14 @@ func (r *TaskReconciler) deletePod(ctx context.Context, t *aliecsv1alpha1.Task, } func (*TaskReconciler) cleargRPC(t *aliecsv1alpha1.Task, log logr.Logger) { - if client, exists := clientsForContainers[t.Name]; exists { + // if client, exists := clientsForContainers[t.Name]; exists { + if client, exists := clientsForContainers.Load(t.Name); exists { log.Info("Cleaning up gRPC connection") - if err := client.Close(); err != nil { + if err := client.(*OccClient).Close(); err != nil { log.Error(err, "Failed to close gRPC client during deletion") } - delete(clientsForContainers, t.Name) + // delete(clientsForContainers, t.Name) + clientsForContainers.Delete(t.Name) log.Info("gRPC cleaned") } } @@ -432,7 +443,7 @@ func (r *TaskReconciler) SetupWithManager(mgr ctrl.Manager) error { }), )). Owns(&v1.Pod{}). - WithOptions(controller.Options{MaxConcurrentReconciles: 1}). + WithOptions(controller.Options{MaxConcurrentReconciles: r.MaxConcurrentReconciles}). Complete(r) } 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