diff --git a/.gitignore b/.gitignore index 6dd29b7..00b41c6 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,2 @@ -bin/ \ No newline at end of file +bin/ +.DS_Store diff --git a/README.md b/README.md index e0d2969..874c21b 100644 --- a/README.md +++ b/README.md @@ -80,7 +80,12 @@ The main version of the plugin supports the creation of docker containers with G CGO_ENABLED=1 GOOS=linux go build -o bin/docker-sd cmd/main.go ``` As you can see, the CGO_ENABLED flag is set to 1, which means that the plugin will be built to allow the GO program to use C code. This is necessary to use the Nvidia GPU libraries. -In the second case, the command is the following: +In the second case, it is first necessary to change the branch of the repository with the following command: + +```bash +git checkout 2-light-version-no-gpu +``` +Then, the command to build the binary executable of the plugin without GPU support is the following: ```bash CGO_ENABLED=0 GOOS=linux go build -o bin/docker-sd cmd/main.go diff --git a/cmd/main.go b/cmd/main.go index 59d0401..710dd4c 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -163,6 +163,9 @@ func main() { log.G(context.Background()).Fatal(err) } + // log the interlinkconfig + log.G(context.Background()).Info("\u2705 InterLinkConfig: ", interLinkConfig) + if interLinkConfig.VerboseLogging { logger.SetLevel(logrus.DebugLevel) } else if interLinkConfig.ErrorsOnlyLogging { @@ -178,17 +181,49 @@ func main() { if availableDinds == "" { availableDinds = "2" } - var dindHandler dindmanager.DindManagerInterface = &dindmanager.DindManager{ - DindList: []dindmanager.DindSpecs{}, - Ctx: ctx, - } availableDindsInt, err := strconv.ParseInt(availableDinds, 10, 8) if err != nil { - log.G(ctx).Info("\u2705 Error parsing availableDinds") + log.G(ctx).Fatal("Error parsing AVAILABLEDINDS: ", err) + } + + // Build the subnet pool from the config (empty slice = no pool). + subnetPool, err := dindmanager.InitSubnetPool(interLinkConfig.DockerNetworkSubnet) + if err != nil { + log.G(ctx).Fatal("Error initialising subnet pool: ", err) + } + if len(subnetPool) > 0 { + log.G(ctx).Info(fmt.Sprintf( + "\u2705 Subnet pool initialised: %d /24 subnets available from %v", + len(subnetPool), interLinkConfig.DockerNetworkSubnet, + )) + } else { + log.G(ctx).Info("\u2705 No DockerNetworkSubnet configured — Docker will assign subnets automatically") } + + var dindHandler dindmanager.DindManagerInterface = &dindmanager.DindManager{ + DindList: []dindmanager.DindSpecs{}, + Ctx: ctx, + FPGAEnabled: interLinkConfig.FPGAEnabled, + XilinxToolsPath: interLinkConfig.XilinxToolsPath, + SubnetPool: subnetPool, + } + + dindHandler.(*dindmanager.DindManager).InitialPoolSz = len(subnetPool) + dindHandler.CleanDindContainers() dindHandler.BuildDindContainers(int8(availableDindsInt)) + // Periodically reclaim orphan DIND networks left behind by failed creates, so + // a single failure cannot permanently exhaust the subnet pool. Only networks + // with no attached container are removed, and never while a build is running. + go func() { + ticker := time.NewTicker(5 * time.Minute) + defer ticker.Stop() + for range ticker.C { + dindHandler.ReapOrphanNetworks() + } + }() + var gpuManager gpustrategies.GPUManagerInterface = &gpustrategies.GPUManager{ GPUSpecsList: []gpustrategies.GPUSpecs{}, Ctx: ctx, @@ -196,17 +231,17 @@ func main() { err = gpuManager.Init() if err != nil { - log.G(ctx).Info("\u274C Init of GPUs failed, error", err) + log.G(ctx).Info("❌ Init of GPUs failed, error", err) } err = gpuManager.Discover() if err != nil { - log.G(ctx).Info("\u274C Discover of GPUs failed, error: ", err) + log.G(ctx).Info("❌ Discover of GPUs failed, error: ", err) } err = gpuManager.Check() if err != nil { - log.G(ctx).Info("\u274C Check of GPUs failed, error: ", err) + log.G(ctx).Info("❌ Check of GPUs failed, error: ", err) } SidecarAPIs := docker.SidecarHandler{ @@ -236,10 +271,12 @@ func main() { log.G(ctx).Info("\u2705 Tracing is disabled") } - if os.Getenv("FPGAENABLED") == "1" { + if interLinkConfig.FPGAEnabled { fpgaManager := &fpgastrategies.FPGAManager{ FPGASpecsList: []fpgastrategies.FPGASpecs{}, Ctx: ctx, + VitisPath: interLinkConfig.VitisPath, + XRTPath: interLinkConfig.XRTPath, } err = fpgaManager.Init() if err != nil { @@ -271,7 +308,7 @@ func main() { // Cleanup the sockfile. c := make(chan os.Signal, 1) - signal.Notify(c, os.Interrupt, syscall.SIGTERM) + signal.Notify(c, os.Interrupt, syscall.SIGTERM, syscall.SIGHUP) go func() { <-c os.Remove(strings.ReplaceAll(interLinkConfig.Socket, "unix://", "")) diff --git a/docker/Dockerfile.sidecar-docker b/docker/Dockerfile.sidecar-docker index de27e6d..42d1ae7 100644 --- a/docker/Dockerfile.sidecar-docker +++ b/docker/Dockerfile.sidecar-docker @@ -7,30 +7,26 @@ RUN CGO_ENABLED=0 GOOS=linux go build -o bin/docker-sidecar cmd/main.go FROM bash:latest as bash-stage # Deploy the application binary into a lean image -FROM docker:24.0-dind-rootless AS build-release-stage +FROM docker:26.1.3-dind AS build-release-stage WORKDIR / +USER root:root + +RUN mkdir -p /sidecar + COPY --from=build-stage /app/bin/docker-sidecar /sidecar/docker-sidecar # adding bash binary to be able to perform commands within the sidecar binary COPY --from=bash-stage /usr/local/bin/bash /bin -USER root:root +ENV INTERLINKCONFIGPATH=/InterLinkConfig.yaml ENV PATH "$PATH:/bin" #creating a simple startup script to start both docker rootless and the sidecar -RUN echo -e '#!/bin/bash\ndockerd-entrypoint.sh & /sidecar/docker-sidecar' > /sidecar/startup-docker.sh +RUN echo -e '#!/bin/bash\ndockerd --mtu=1350 & /sidecar/docker-sidecar' > /sidecar/startup-docker.sh RUN chmod +x /sidecar/startup-docker.sh -RUN chmod -R 777 /sidecar - -ENV INTERLINKCONFIGPATH=/InterLinkConfig.yaml - -USER 1000:1000 - -#setting up the path for the docker daemon -ENV DOCKER_HOST=unix:///run/user/1000/docker.sock WORKDIR /sidecar diff --git a/pkg/common/types.go b/pkg/common/types.go index 49b4fef..4912358 100644 --- a/pkg/common/types.go +++ b/pkg/common/types.go @@ -54,31 +54,32 @@ type RetrievedContainer struct { // InterLinkConfig holds the whole configuration type InterLinkConfig struct { - VKConfigPath string `yaml:"VKConfigPath"` - VKTokenFile string `yaml:"VKTokenFile"` - Interlinkurl string `yaml:"InterlinkURL"` - Sidecarurl string `yaml:"SidecarURL"` - Sbatchpath string `yaml:"SbatchPath"` - Scancelpath string `yaml:"ScancelPath"` - Squeuepath string `yaml:"SqueuePath"` - Interlinkport string `yaml:"InterlinkPort"` - Socket string `yaml:"Socket"` - Sidecarport string `yaml:"SidecarPort"` - Commandprefix string `yaml:"CommandPrefix"` - ExportPodData bool `yaml:"ExportPodData"` - DataRootFolder string `yaml:"DataRootFolder"` - ServiceAccount string `yaml:"ServiceAccount"` - Namespace string `yaml:"Namespace"` - Tsocks bool `yaml:"Tsocks"` - Tsockspath string `yaml:"TsocksPath"` - Tsocksconfig string `yaml:"TsocksConfig"` - Tsockslogin string `yaml:"TsocksLoginNode"` - BashPath string `yaml:"BashPath"` - VerboseLogging bool `yaml:"VerboseLogging"` - ErrorsOnlyLogging bool `yaml:"ErrorsOnlyLogging"` - PodIP string `yaml:"PodIP"` - SingularityPrefix string `yaml:"SingularityPrefix"` - set bool + VKConfigPath string `yaml:"VKConfigPath"` + VKTokenFile string `yaml:"VKTokenFile"` + Interlinkurl string `yaml:"InterlinkURL"` + Sidecarurl string `yaml:"SidecarURL"` + Sbatchpath string `yaml:"SbatchPath"` + Scancelpath string `yaml:"ScancelPath"` + Squeuepath string `yaml:"SqueuePath"` + Interlinkport string `yaml:"InterlinkPort"` + Socket string `yaml:"Socket"` + Sidecarport string `yaml:"SidecarPort"` + Commandprefix string `yaml:"CommandPrefix"` + ExportPodData bool `yaml:"ExportPodData"` + DataRootFolder string `yaml:"DataRootFolder"` + ServiceAccount string `yaml:"ServiceAccount"` + Namespace string `yaml:"Namespace"` + Tsocks bool `yaml:"Tsocks"` + Tsockspath string `yaml:"TsocksPath"` + Tsocksconfig string `yaml:"TsocksConfig"` + Tsockslogin string `yaml:"TsocksLoginNode"` + BashPath string `yaml:"BashPath"` + VerboseLogging bool `yaml:"VerboseLogging"` + ErrorsOnlyLogging bool `yaml:"ErrorsOnlyLogging"` + PodIP string `yaml:"PodIP"` + SingularityPrefix string `yaml:"SingularityPrefix"` + DockerNetworkSubnet []string `yaml:"DockerNetworkSubnet"` + set bool } // ContainerLogOpts is a struct in which it is possible to specify options to retrieve logs from the sidecar diff --git a/pkg/docker/Create.go b/pkg/docker/Create.go index d9e64b6..b56b2d0 100644 --- a/pkg/docker/Create.go +++ b/pkg/docker/Create.go @@ -1,6 +1,7 @@ package docker import ( + "context" "encoding/json" "fmt" "io" @@ -10,6 +11,8 @@ import ( "strings" "time" + OSexec "os/exec" + exec "github.com/alexellis/go-execute/pkg/v1" "github.com/containerd/containerd/log" v1 "k8s.io/api/core/v1" @@ -17,7 +20,6 @@ import ( "errors" commonIL "github.com/interlink-hq/interlink/pkg/interlink" - "github.com/intertwin-eu/interlink-docker-plugin/pkg/docker/dindmanager" "path/filepath" @@ -26,11 +28,49 @@ import ( trace "go.opentelemetry.io/otel/trace" ) +const ( + // meshReadyTimeoutSeconds bounds how long containers_command.sh waits for the + // mesh_ready sentinel written by the network-overlay container. + meshReadyTimeoutSeconds = 300 + + // containersCommandTimeout bounds the whole containers_command.sh execution + // inside the DIND container. It must be larger than meshReadyTimeoutSeconds so + // the script gets the chance to report the mesh timeout itself. + containersCommandTimeout = 10 * time.Minute +) + +// selectNetworkOverlay splits containers into the network-overlay container and +// the workload containers that have to wait for the mesh sentinel it writes. +// ok is false when overlayName is empty (no overlay was built for this pod) or +// when no container carries that name, in which case overlay is the zero value +// and workload holds every container unchanged. +// +// The overlay is prepended to the list, so it is normally at index 0. Taking it +// from there unconditionally was the bug: the condition that selected the mesh +// startup path (the pod carries a mesh annotation) is weaker than the conditions +// under which the overlay is actually built, so when the two disagreed an +// ordinary workload container was started as the overlay and every other +// container waited on a sentinel nothing would ever write. It also panicked on +// an empty container list. Matching by name makes the two impossible to confuse. +func selectNetworkOverlay(containers []DockerRunStruct, overlayName string) (overlay DockerRunStruct, workload []DockerRunStruct, ok bool) { + if overlayName == "" { + return DockerRunStruct{}, containers, false + } + for idx, c := range containers { + if c.Name != overlayName { + continue + } + workload = make([]DockerRunStruct, 0, len(containers)-1) + workload = append(workload, containers[:idx]...) + workload = append(workload, containers[idx+1:]...) + return c, workload, true + } + return DockerRunStruct{}, containers, false +} + func (h *SidecarHandler) prepareDockerRuns(podData commonIL.RetrievedPodData, w http.ResponseWriter) ([]DockerRunStruct, error) { var dockerRunStructs []DockerRunStruct - var gpuArgs string = "" - var fpgaArgs string = "" podUID := string(podData.Pod.UID) podNamespace := string(podData.Pod.Namespace) @@ -82,6 +122,14 @@ func (h *SidecarHandler) prepareDockerRuns(podData commonIL.RetrievedPodData, w for _, container := range containers { + // envVars, fpgaArgs and gpuArgs MUST be scoped to the single container: + // when they were declared outside this loop every container inherited the + // -e/-v flags, --device entries and GPU assignment of the containers + // processed before it. + var envVars string = "" + var fpgaArgs string = "" + var gpuArgs string = "" + containerName := podNamespace + "-" + podUID + "-" + container.Name var isGpuRequested bool = false @@ -94,10 +142,16 @@ func (h *SidecarHandler) prepareDockerRuns(podData commonIL.RetrievedPodData, w // if the container is requesting 0 GPU, skip the GPU assignment if numGpusRequested == 0 { - log.G(h.Ctx).Info("\u2705 Container " + containerName + " is not requesting a GPU") + log.G(h.Ctx).Info("✅ Container " + containerName + " is not requesting a GPU") } else { - log.G(h.Ctx).Info("\u2705 Container " + containerName + " is requesting " + val.String() + " GPU") + if h.GpuManager == nil { + log.G(h.Ctx).Error("❌ [CREATE CALL] GPU Manager is not initialized") + HandleErrorAndRemoveData(h, w, "GPU Manager is not initialized", errors.New("GPU Manager is not initialized"), podNamespace, podUID) + return dockerRunStructs, errors.New("GPU Manager is not initialized") + } + + log.G(h.Ctx).Info("✅ Container " + containerName + " is requesting " + val.String() + " GPU") isGpuRequested = true @@ -127,7 +181,6 @@ func (h *SidecarHandler) prepareDockerRuns(podData commonIL.RetrievedPodData, w additionalGpuArgs = append(additionalGpuArgs, "--runtime=nvidia -e NVIDIA_VISIBLE_DEVICES="+gpuUUIDs) gpuArgs = "--runtime=nvidia -e NVIDIA_VISIBLE_DEVICES=" + gpuUUIDs } - } if val, ok := container.Resources.Limits["xilinx.com/fpga"]; ok { @@ -136,19 +189,39 @@ func (h *SidecarHandler) prepareDockerRuns(podData commonIL.RetrievedPodData, w log.G(h.Ctx).Info("\u2705 Container " + containerName + " is not requesting a FPGA") } else { + if h.FPGAManager == nil { + log.G(h.Ctx).Error("\u274C [CREATE CALL] FPGA Manager is not initialized") + HandleErrorAndRemoveData(h, w, "FPGA Manager is not initialized", errors.New("FPGA Manager is not initialized"), podNamespace, podUID) + return dockerRunStructs, errors.New("FPGA Manager is not initialized") + } + isFPGARequested = true log.G(h.Ctx).Info("\u2705 Container " + containerName + " is requesting " + strconv.Itoa(int(numFPGAsRequested)) + " FPGA(s)") numFPGAsRequestedInt := int(numFPGAsRequested) _, err := h.FPGAManager.GetAvailableFPGAs(numFPGAsRequestedInt) + log.G(h.Ctx).Info("\u2705 [CREATE CALL] Retrieved available FPGAs") if err != nil { + // log the err + log.G(h.Ctx).Error("\u274C [CREATE CALL] Error retrieving available FPGAs") HandleErrorAndRemoveData(h, w, "An error occurred during the request of available FPGAs", err, podNamespace, podUID) return dockerRunStructs, errors.New("An error occurred during the request of available FPGAs") } + + log.G(h.Ctx).Info("\u2705 [CREATE CALL] ********* BEFORE Requested FPGAs are available") + assignedFPGAs, err := h.FPGAManager.GetAndAssignAvailableFPGAs(numFPGAsRequestedInt, containerName) + log.G(h.Ctx).Info("\u2705 [CREATE CALL] ********* AFTER Requested FPGAs are available") + + // log the assigned FPGAs + log.G(h.Ctx).Info("\u2705 [CREATE CALL] Assigned FPGAs: ") + for _, fpgaSpec := range assignedFPGAs { + log.G(h.Ctx).Info("\u2705 [CREATE CALL] " + fpgaSpec.DeviceToMount) + } if err != nil { - HandleErrorAndRemoveData(h, w, "An error occurred during request of get and assign of an available GPU", err, podNamespace, podUID) - return dockerRunStructs, errors.New("An error occurred during request of get and assign of an available GPU") + log.G(h.Ctx).Error("\u274C [CREATE CALL] Error during request of get and assign of an available FPGA") + HandleErrorAndRemoveData(h, w, "An error occurred during request of get and assign of an available FPGA", err, podNamespace, podUID) + return dockerRunStructs, errors.New("An error occurred during request of get and assign of an available FPGA") } for _, fpgaSpec := range assignedFPGAs { fpgaArgs += " --device=" + fpgaSpec.DeviceToMount + ":" + fpgaSpec.DeviceToMount @@ -156,7 +229,6 @@ func (h *SidecarHandler) prepareDockerRuns(podData commonIL.RetrievedPodData, w } } - var envVars string for _, envVar := range container.Env { if envVar.Value != "" { value := envVar.Value @@ -200,6 +272,13 @@ func (h *SidecarHandler) prepareDockerRuns(podData commonIL.RetrievedPodData, w } } + // if FPGA is requested, mount in read mode the Xilinx tools path in the container + if isFPGARequested { + envVars += " -v " + h.Config.XilinxToolsPath + ":" + h.Config.XilinxToolsPath + ":ro" + } + + log.G(h.Ctx).Info("\u2705 [POD FLOW] Before creating run command") + //envVars += " --network=host" cmd := []string{"run", "--user", "root", "-d", "--name", containerName} @@ -220,11 +299,12 @@ func (h *SidecarHandler) prepareDockerRuns(podData commonIL.RetrievedPodData, w var additionalPortArgs []string for _, port := range container.Ports { - if port.HostPort != 0 { - additionalPortArgs = append(additionalPortArgs, "-p", strconv.Itoa(int(port.HostPort))+":"+strconv.Itoa(int(port.ContainerPort))) - } + log.G(h.Ctx).Info("\u2705 [POD FLOW] Container port: " + strconv.Itoa(int(port.ContainerPort)) + " Protocol: " + string(port.Protocol) + " HostPort: " + strconv.Itoa(int(port.HostPort))) + additionalPortArgs = append(additionalPortArgs, "-p", strconv.Itoa(int(port.ContainerPort))+":"+strconv.Itoa(int(port.ContainerPort))) } + log.G(h.Ctx).Info("\u2705 [POD FLOW] Additional port arguments for container " + containerName + ": " + strings.Join(additionalPortArgs, " ")) + cmd = append(cmd, additionalPortArgs...) mounts, err := prepareMounts(h.Ctx, h.Config, podData, container) @@ -304,39 +384,12 @@ func (h *SidecarHandler) CreateHandler(w http.ResponseWriter, r *http.Request) { attribute.Int64("start.timestamp", start), )) - // create bool variable to set if a new dind container has to be created - newDindContainerCreated := false - - // get a dind container ID from dind manager of the sidecard handler - dindContainerID, err := h.DindManager.GetAvailableDind() - if err != nil { - - log.G(h.Ctx).Info("\u2705 [POD FLOW] No available DIND container found, creating a new one") - - h.DindManager.BuildDindContainers(1) - dindContainerID, err = h.DindManager.GetAvailableDind() - if err != nil { - HandleErrorAndRemoveData(h, w, "During creation of new DIND container, an error occurred during the request of get available DIND container", err, "", "") - return - } - newDindContainerCreated = true - } - - // remove the dind container from the list of available dind containers - err = h.DindManager.SetDindUnavailable(dindContainerID) - if err != nil { - HandleErrorAndRemoveData(h, w, "An error occurred during the removal of the DIND container from the list of available DIND containers", err, "", "") - return - } - - if !newDindContainerCreated { - // create a new dind container in background - go h.DindManager.BuildDindContainers(1) - } - - //var execReturn exec.ExecResult statusCode := http.StatusOK + // Read and parse the request body FIRST, so the pod UID is known before a DIND + // container is claimed. Claiming with the pod UID (ClaimAvailableDind) makes + // the assignment atomic \u2014 Available=false and PodUID are set together \u2014 so a + // failure at any later point can always find and clean up the right DIND. bodyBytes, err := io.ReadAll(r.Body) if err != nil { HandleErrorAndRemoveData(h, w, "An error occurred during read of body request for pod creation", err, "", "") @@ -344,20 +397,41 @@ func (h *SidecarHandler) CreateHandler(w http.ResponseWriter, r *http.Request) { } var req commonIL.RetrievedPodData - err = json.Unmarshal(bodyBytes, &req) - - if err != nil { + if err = json.Unmarshal(bodyBytes, &req); err != nil { HandleErrorAndRemoveData(h, w, "An error occurred during json unmarshal of data from pod creation request", err, "", "") return } - wd, err := os.Getwd() + log.G(h.Ctx).Info("\u2705 [POD FLOW] Request data unmarshalled successfully") + + podNamespace := string(req.Pod.Namespace) + podUID := string(req.Pod.UID) + + // Atomically claim a DIND container for this pod. + newDindContainerCreated := false + dindContainerID, err := h.DindManager.ClaimAvailableDind(podUID) if err != nil { - HandleErrorAndRemoveData(h, w, "Unable to get current working directory", err, "", "") - return + + log.G(h.Ctx).Info("\u2705 [POD FLOW] No available DIND container found, creating a new one") + + // The build error must be reported: without it the only thing reaching the + // caller is the generic "no available DIND container" from the claim below. + if buildErr := h.DindManager.BuildDindContainers(1); buildErr != nil { + HandleErrorAndRemoveData(h, w, "An error occurred during the creation of a new DIND container", buildErr, podNamespace, podUID) + return + } + dindContainerID, err = h.DindManager.ClaimAvailableDind(podUID) + if err != nil { + HandleErrorAndRemoveData(h, w, "During creation of new DIND container, an error occurred during the request of get available DIND container", err, podNamespace, podUID) + return + } + newDindContainerCreated = true } - log.G(h.Ctx).Info("\u2705 [POD FLOW] Request data unmarshalled successfully and current working directory detected") + if !newDindContainerCreated { + // replenish the pool in the background since we consumed a pre-warmed one + go h.DindManager.BuildDindContainers(1) + } var newReq []commonIL.RetrievedPodData newReq = []commonIL.RetrievedPodData{req} @@ -367,7 +441,11 @@ func (h *SidecarHandler) CreateHandler(w http.ResponseWriter, r *http.Request) { podUID := string(data.Pod.UID) podNamespace := string(data.Pod.Namespace) - podDirectoryPath := filepath.Join(wd, h.Config.DataRootFolder+"/"+podNamespace+"-"+podUID) + podDirectoryPath := filepath.Join(h.Config.DataRootFolder, podNamespace+"-"+podUID) + + // Sentinel file written by mesh.sh once network setup is complete. + // containers_command.sh polls for this file before starting workload containers. + meshReadyFile := filepath.Join(podDirectoryPath, "mesh_ready") // log the pod specifics log.G(h.Ctx).Info(fmt.Sprintf("\u2705 [POD FLOW] Pod specs: %+v", data.Pod)) @@ -382,7 +460,7 @@ func (h *SidecarHandler) CreateHandler(w http.ResponseWriter, r *http.Request) { if _, err := os.Stat(podDirectoryPath); os.IsNotExist(err) { err = os.MkdirAll(podDirectoryPath, os.ModePerm) if err != nil { - HandleErrorAndRemoveData(h, w, "An error occurred during the creation of the pod directory", err, "", "") + HandleErrorAndRemoveData(h, w, "An error occurred during the creation of the pod directory", err, podNamespace, podUID) return } } @@ -390,10 +468,18 @@ func (h *SidecarHandler) CreateHandler(w http.ResponseWriter, r *http.Request) { // call prepareDockerRuns to get the DockerRunStruct array dockerRunStructs, err := h.prepareDockerRuns(data, w) if err != nil { - HandleErrorAndRemoveData(h, w, "An error occurred during preparing of docker run commmands", err, "", "") + HandleErrorAndRemoveData(h, w, "An error occurred during preparing of docker run commmands", err, podNamespace, podUID) return } + // Name of the network-overlay container, set below only if one is actually + // built and prepended to dockerRunStructs. Everything downstream keys off + // this instead of re-testing the annotation: the annotation check and the + // overlay creation do not have the same preconditions, and assuming they + // agree is what made the startup script treat an ordinary workload container + // as the overlay. + networkOverlayName := "" + if preExecAnnotations, ok := data.Pod.Annotations["slurm-job.vk.io/pre-exec"]; ok { if strings.Contains(preExecAnnotations, "cat <<'EOFMESH' > $TMPDIR/mesh.sh") { meshScript, err := extractHeredoc(preExecAnnotations, "EOFMESH") @@ -430,14 +516,17 @@ func (h *SidecarHandler) CreateHandler(w http.ResponseWriter, r *http.Request) { // Remove the slirp4netns execution at the end of inner script innerScript = removeSlirp4netnsExecution(innerScript) - // Replace command execution with sleep infinity - innerScript = strings.Replace(innerScript, "$@", "sleep infinity", -1) + // Replace command execution with a sentinel touch followed by sleep infinity. + // The sentinel file signals to containers_command.sh that network setup is done. + innerScript = strings.Replace(innerScript, "$@", "touch "+meshReadyFile+" && sleep infinity", -1) // Build the complete script with correct order meshScript = `#!/bin/bash set -e set -m +sleep 20s + export PATH=$PATH:$PWD:/usr/sbin:/sbin # Set up temporary directory @@ -453,10 +542,10 @@ cd $TMPDIR ` + innerScript } else { - // Fallback: just clean up the outer script + // Fallback: just clean up the outer script, still touch the sentinel before sleeping. meshScript = removeSlirp4netnsDownload(meshScript) meshScript = removeUnshareWrapper(meshScript) - meshScript = strings.Replace(meshScript, "$@", "sleep infinity", -1) + meshScript = strings.Replace(meshScript, "$@", "touch "+meshReadyFile+" && sleep infinity", -1) meshScript = removeSlirp4netnsExecution(meshScript) } @@ -501,6 +590,8 @@ cd $TMPDIR FpgaArgs: "", }}, dockerRunStructs...) + networkOverlayName = networkContainerName + log.G(h.Ctx).Info("✅ [POD FLOW] Network overlay container prepared: " + networkContainerName) } else { log.G(h.Ctx).Error("❌ [POD FLOW] Failed to extract mesh.sh script from annotation") @@ -527,13 +618,6 @@ cd $TMPDIR } } - // set the podUID to the dind container - err = h.DindManager.SetPodUIDToDind(dindContainerID, podUID) - if err != nil { - HandleErrorAndRemoveData(h, w, "An error occurred during the setting of the pod UID to the DIND container", err, "", "") - return - } - // run the docker command to rename the container to the pod UID shell := exec.ExecTask{ Command: "docker", @@ -543,7 +627,7 @@ cd $TMPDIR _, err = shell.Execute() if err != nil { - HandleErrorAndRemoveData(h, w, "An error occurred during the rename of the DIND container", err, "", "") + HandleErrorAndRemoveData(h, w, "An error occurred during the rename of the DIND container", err, podNamespace, podUID) return } @@ -568,40 +652,40 @@ cd $TMPDIR } // Create DNS configuration script for DIND - dnsConfigScript := `#!/bin/sh -set -e - -# Backup original resolv.conf -cp /etc/resolv.conf /etc/resolv.conf.backup 2>/dev/null || true - -# Create new resolv.conf with cluster DNS -cat > /etc/resolv.conf << EOF -nameserver 8.8.8.8 -search ` + dnsSearch + ` -EOF - -echo "DNS configured for cluster connectivity" -` - - // Write DNS config script to pod directory - dnsScriptPath := filepath.Join(podDirectoryPath, "configure-dns.sh") - err = os.WriteFile(dnsScriptPath, []byte(dnsConfigScript), 0755) - if err != nil { - log.G(h.Ctx).Warning("⚠️ Failed to create DNS config script: " + err.Error()) - } else { - // Execute DNS configuration on DIND container - dnsExecCmd := exec.ExecTask{ - Command: "docker", - Args: []string{"exec", string(data.Pod.UID) + "_dind", "sh", dnsScriptPath}, - Shell: true, - } - _, err = dnsExecCmd.Execute() - if err != nil { - log.G(h.Ctx).Warning("⚠️ Failed to configure DNS on DIND container: " + err.Error()) - } else { - log.G(h.Ctx).Info("✅ [POD FLOW] DNS configured on DIND container successfully with NS: " + dnsNameserver + ", Search: " + dnsSearch) - } - } + /* dnsConfigScript := `#!/bin/sh + set -e + + # Backup original resolv.conf + cp /etc/resolv.conf /etc/resolv.conf.backup 2>/dev/null || true + + # Create new resolv.conf with cluster DNS + cat > /etc/resolv.conf << EOF + nameserver 8.8.8.8 + search ` + dnsSearch + ` + EOF + + echo "DNS configured for cluster connectivity" + ` + + // Write DNS config script to pod directory + dnsScriptPath := filepath.Join(podDirectoryPath, "configure-dns.sh") + err = os.WriteFile(dnsScriptPath, []byte(dnsConfigScript), 0755) + if err != nil { + log.G(h.Ctx).Warning("⚠️ Failed to create DNS config script: " + err.Error()) + } else { + // Execute DNS configuration on DIND container + dnsExecCmd := exec.ExecTask{ + Command: "docker", + Args: []string{"exec", string(data.Pod.UID) + "_dind", "sh", dnsScriptPath}, + Shell: true, + } + _, err = dnsExecCmd.Execute() + if err != nil { + log.G(h.Ctx).Warning("⚠️ Failed to configure DNS on DIND container: " + err.Error()) + } else { + log.G(h.Ctx).Info("✅ [POD FLOW] DNS configured on DIND container successfully with NS: " + dnsNameserver + ", Search: " + dnsSearch) + } + } */ } } @@ -609,7 +693,7 @@ echo "DNS configured for cluster connectivity" createResponseBytes, err := json.Marshal(createResponse) if err != nil { statusCode = http.StatusInternalServerError - HandleErrorAndRemoveData(h, w, "An error occurred during the json marshal of the returned JID", err, "", "") + HandleErrorAndRemoveData(h, w, "An error occurred during the json marshal of the returned JID", err, podNamespace, podUID) return } @@ -633,6 +717,17 @@ echo "DNS configured for cluster connectivity" span.End() go func() { + // The HTTP response has already been sent at this point, so nothing below + // may touch w. A panic here would also not be recovered by net/http (it is + // a bare goroutine, not the handler), and would take the whole plugin down + // together with the bookkeeping of every other running pod. + defer func() { + if r := recover(); r != nil { + log.G(h.Ctx).Errorf("❌ [POD FLOW] panic while creating containers for pod %s: %v", podUID, r) + cleanupPodData(h, "Panic during asynchronous container creation", + fmt.Errorf("panic: %v", r), podNamespace, podUID) + } + }() if len(initContainers) > 0 { @@ -661,7 +756,7 @@ echo "DNS configured for cluster connectivity" _, err := shell.Execute() if err != nil { - HandleErrorAndRemoveData(h, w, "An error occurred during the exec of the init container command", err, "", "") + cleanupPodData(h, "An error occurred during the exec of the init container command", err, podNamespace, podUID) return } @@ -674,7 +769,7 @@ echo "DNS configured for cluster connectivity" statusReturn, err := shell.Execute() if err != nil { - HandleErrorAndRemoveData(h, w, "An error occurred during inspect of init container", err, "", "") + cleanupPodData(h, "An error occurred during inspect of init container", err, podNamespace, podUID) return } @@ -696,7 +791,14 @@ echo "DNS configured for cluster connectivity" // create a file called containers_command.sh and write the containers commands to it, use WriteFile function containersCommand := "#!/bin/sh\n" - if isMeshScriptPresent { + networkOverlay, workloadContainers, hasOverlay := selectNetworkOverlay(containers, networkOverlayName) + + if isMeshScriptPresent && !hasOverlay { + log.G(h.Ctx).Warning("⚠️ [POD FLOW] Pod " + podUID + " carries a mesh pre-exec annotation but no network-overlay " + + "container was created; starting its containers directly, without cluster network setup or cluster DNS") + } + + if hasOverlay { dnsNameserver := h.FinalDNSNameserver dnsSearch := h.FinalDNSSearch @@ -708,12 +810,31 @@ echo "DNS configured for cluster connectivity" dnsSearch = "default.svc.cluster.local svc.cluster.local cluster.local" // Fallback } - // Add a delay to ensure network container is ready - containersCommand += "echo 'Waiting for network overlay to be ready...'\n" - containersCommand += "sleep 10\n" - containersCommand += "echo 'Starting containers...'\n\n" - - for _, container := range containers { + containersCommand += "# Start network overlay container first\n" + containersCommand += networkOverlay.Command + "\n\n" + + // Poll for the sentinel file written by mesh.sh once network setup is + // complete. The wait MUST be bounded: mesh.sh downloads binaries and + // brings up WireGuard, and when any of that fails the sentinel is never + // written. An unbounded loop left the docker exec below blocked with no + // diagnostics, the pod stuck in Waiting and its DIND claimed until a + // human deleted the pod. Timing out instead fails the pod loudly and + // lets the deferred cleanup reclaim the DIND. + containersCommand += "echo 'Waiting for network overlay to be ready (timeout " + strconv.Itoa(meshReadyTimeoutSeconds) + "s)...'\n" + containersCommand += "meshWaited=0\n" + containersCommand += "while [ ! -f " + meshReadyFile + " ]; do\n" + containersCommand += " if [ \"$meshWaited\" -ge " + strconv.Itoa(meshReadyTimeoutSeconds) + " ]; then\n" + containersCommand += " echo 'ERROR: network overlay did not become ready after " + strconv.Itoa(meshReadyTimeoutSeconds) + "s; aborting container startup'\n" + containersCommand += " docker logs " + networkOverlay.Name + " 2>&1 | tail -n 50\n" + containersCommand += " exit 1\n" + containersCommand += " fi\n" + containersCommand += " echo \"Network not ready yet, waiting 2s (${meshWaited}s elapsed)...\"\n" + containersCommand += " sleep 2\n" + containersCommand += " meshWaited=$((meshWaited+2))\n" + containersCommand += "done\n" + containersCommand += "echo 'Network overlay is ready (sentinel file found), starting containers...'\n\n" + + for _, container := range workloadContainers { containersCommand += "# Start container: " + container.Name + "\n" containersCommand += container.Command + "\n" containersCommand += "sleep 2\n" @@ -724,37 +845,44 @@ echo "DNS configured for cluster connectivity" containersCommand += "cp /etc/resolv.conf /etc/resolv.conf.backup 2>/dev/null || true\n" containersCommand += "cat > /etc/resolv.conf << EOF\n" containersCommand += "nameserver " + dnsNameserver + "\n" + containersCommand += "nameserver 8.8.8.8 \n" containersCommand += "search " + dnsSearch + "\n" containersCommand += "EOF\n" containersCommand += "' || echo 'Warning: Could not configure DNS for " + container.Name + "'\n" containersCommand += "echo 'DNS configured for container: " + container.Name + "'\n\n" } - } - - for _, container := range containers { - containersCommand += container.Command + "\n" - containersCommand += "sleep 30\n" + } else { + for _, container := range containers { + containersCommand += container.Command + "\n" + containersCommand += "sleep 1\n" + } } err = os.WriteFile(podDirectoryPath+"/containers_command.sh", []byte(containersCommand), 0644) if err != nil { log.G(h.Ctx).Error("\u274C [POD FLOW] Error writing containers command script: " + err.Error()) - HandleErrorAndRemoveData(h, w, "An error occurred during the creation of the container commands script.", err, "", "") + cleanupPodData(h, "An error occurred during the creation of the container commands script.", err, podNamespace, podUID) return } log.G(h.Ctx).Info("\u2705 [POD FLOW] Containers commands written to the script file") - shell = exec.ExecTask{ - Command: "docker", - Args: []string{"exec", string(data.Pod.UID) + "_dind", "/bin/sh", podDirectoryPath + "/containers_command.sh"}, - } + // Bound the exec as well, so a script that blocks for any other reason + // cannot leak this goroutine and keep the DIND claimed forever. Unlike + // go-execute, OSexec.CommandContext can be cancelled. + execCtx, cancelExec := context.WithTimeout(h.Ctx, containersCommandTimeout) + defer cancelExec() - log.G(h.Ctx).Info("\u2705 [POD FLOW] Executing containers creation script inside DIND container; command to execute: docker " + strings.Join(shell.Args, " ")) + execArgs := []string{"exec", string(data.Pod.UID) + "_dind", "/bin/sh", podDirectoryPath + "/containers_command.sh"} + log.G(h.Ctx).Info("\u2705 [POD FLOW] Executing containers creation script inside DIND container; command to execute: docker " + strings.Join(execArgs, " ")) - _, err = shell.Execute() + scriptOutput, err := OSexec.CommandContext(execCtx, "docker", execArgs...).CombinedOutput() if err != nil { + if execCtx.Err() == context.DeadlineExceeded { + err = fmt.Errorf("containers command script timed out after %s: %w", containersCommandTimeout, err) + } log.G(h.Ctx).Error("\u274C [POD FLOW] Error executing containers command script: " + err.Error()) - HandleErrorAndRemoveData(h, w, "An error occurred during the execution of the container command script", err, "", "") + log.G(h.Ctx).Error("\u274C [POD FLOW] Script output: " + strings.TrimSpace(string(scriptOutput))) + cleanupPodData(h, "An error occurred during the execution of the container command script", err, podNamespace, podUID) return } @@ -765,43 +893,65 @@ echo "DNS configured for cluster connectivity" } +// HandleErrorAndRemoveData reports the failure on the HTTP response and then +// reclaims everything provisioned for the pod. It may only be called from the +// request goroutine, before the response has been written; anything running +// after the handler returned must call cleanupPodData directly, because writing +// to a ResponseWriter after its handler returned races with the http server and +// can corrupt the next response on a keep-alive connection. func HandleErrorAndRemoveData(h *SidecarHandler, w http.ResponseWriter, s string, err error, podNamespace string, podUID string) { - log.G(h.Ctx).Error(err) - log.G(h.Ctx).Info("\u274C Error description: " + s) w.WriteHeader(http.StatusInternalServerError) w.Write([]byte("Some errors occurred while creating container. Check Docker Sidecar's logs")) + cleanupPodData(h, s, err, podNamespace, podUID) +} + +// cleanupPodData tears down the pod data directory and the DIND container, +// network and subnet claimed for the pod. It never touches the HTTP response, so +// it is safe to call from the asynchronous container-creation goroutine. +func cleanupPodData(h *SidecarHandler, s string, err error, podNamespace string, podUID string) { + if err != nil { + log.G(h.Ctx).Error(err) + } + log.G(h.Ctx).Info("\u274C Error description: " + s) if podNamespace != "" && podUID != "" { os.RemoveAll(h.Config.DataRootFolder + podNamespace + "-" + podUID) } - dindSpec := dindmanager.DindSpecs{} - dindSpec, err = h.DindManager.GetDindFromPodUID(podUID) - if err != nil { - log.G(h.Ctx).Error("\u274C [CREATE CALL] Error retrieving DindSpecs, maybe the Dind container has already been deleted") - } else { - log.G(h.Ctx).Info("\u2705 [CREATE CALL] Retrieved DindSpecs: " + dindSpec.DindID + " " + dindSpec.PodUID + " " + dindSpec.DindNetworkID + " ") + // Without a pod UID there is no DIND to reconcile (e.g. the body was never + // parsed). Bail out here rather than matching an unrelated DIND by "". + if podUID == "" { + return + } - // log the retrieved dindSpec - log.G(h.Ctx).Info("\u2705 [CREATE CALL] Retrieved DindSpecs: " + dindSpec.DindID + " " + dindSpec.PodUID + " " + dindSpec.DindNetworkID + " ") + dindSpec, lookupErr := h.DindManager.GetDindFromPodUID(podUID) + if lookupErr != nil { + log.G(h.Ctx).Info("\u2139\uFE0F [CREATE CALL] No DIND container assigned to pod " + podUID + " to clean up") + return + } - cmd := []string{"network", "rm", dindSpec.DindNetworkID} - shell := exec.ExecTask{ - Command: "docker", - Args: cmd, - Shell: true, + log.G(h.Ctx).Info("\u2705 [CREATE CALL] Cleaning up DIND for pod " + podUID + ": " + dindSpec.DindID + " (network " + dindSpec.DindNetworkID + ")") + + // Force-remove the DIND container FIRST. While it is running it holds an + // endpoint on its bridge network, so the network cannot be removed and its + // /24 subnet stays effectively allocated \u2014 which then causes the next create + // drawing that subnet from the pool to fail with an address overlap. The + // container may be named after the pod UID (after the rename step) or after + // its original build UID (before it); remove both, ignoring "no such + // container" errors. + for _, name := range []string{podUID + "_dind", dindSpec.DindID} { + if name == "" { + continue } - execReturn, _ := shell.Execute() - execReturn.Stdout = strings.ReplaceAll(execReturn.Stdout, "\n", "") - if execReturn.Stderr != "" { - log.G(h.Ctx).Error("\u274C [CREATE CALL] Error deleting network " + dindSpec.DindNetworkID) - } else { - log.G(h.Ctx).Info("\u2705 [CREATE CALL] Deleted network " + dindSpec.DindNetworkID) - } - // set the dind available again - err = h.DindManager.RemoveDindFromList(dindSpec.PodUID) - if err != nil { - log.G(h.Ctx).Error("\u274C [CREATE CALL] Error setting DIND container available") + rm := exec.ExecTask{Command: "docker", Args: []string{"rm", "-f", name}, Shell: true} + if execReturn, rmErr := rm.Execute(); rmErr != nil || execReturn.ExitCode != 0 { + log.G(h.Ctx).Warning("\u26A0\uFE0F [CREATE CALL] Could not remove DIND container " + name + ": " + strings.TrimSpace(execReturn.Stderr)) } } + + // Now that the container is gone, RemoveDindFromList can actually delete the + // bridge network and return the subnet to the pool. + if remErr := h.DindManager.RemoveDindFromList(dindSpec.PodUID); remErr != nil { + log.G(h.Ctx).Error("\u274C [CREATE CALL] Error removing DIND from list: " + remErr.Error()) + } } diff --git a/pkg/docker/Create_test.go b/pkg/docker/Create_test.go new file mode 100644 index 0000000..dbd7d9c --- /dev/null +++ b/pkg/docker/Create_test.go @@ -0,0 +1,123 @@ +package docker + +import "testing" + +func names(structs []DockerRunStruct) []string { + out := make([]string, 0, len(structs)) + for _, s := range structs { + out = append(out, s.Name) + } + return out +} + +func equal(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} + +func TestSelectNetworkOverlay(t *testing.T) { + overlay := DockerRunStruct{Name: "ns-uid-network-overlay"} + appC := DockerRunStruct{Name: "ns-uid-app"} + sidecar := DockerRunStruct{Name: "ns-uid-sidecar"} + + tests := []struct { + name string + containers []DockerRunStruct + overlayName string + wantOK bool + wantOverlay string + wantWorkload []string + }{ + { + name: "overlay first, as prepended", + containers: []DockerRunStruct{overlay, appC, sidecar}, + overlayName: overlay.Name, + wantOK: true, + wantOverlay: overlay.Name, + wantWorkload: []string{appC.Name, sidecar.Name}, + }, + { + name: "overlay not at index 0 is still found, order preserved", + containers: []DockerRunStruct{appC, overlay, sidecar}, + overlayName: overlay.Name, + wantOK: true, + wantOverlay: overlay.Name, + wantWorkload: []string{appC.Name, sidecar.Name}, + }, + { + name: "overlay is the only container", + containers: []DockerRunStruct{overlay}, + overlayName: overlay.Name, + wantOK: true, + wantOverlay: overlay.Name, + wantWorkload: []string{}, + }, + { + // The regression: a mesh annotation was present but no overlay was built. + // containers[0] used to be started as the overlay, and every remaining + // container then waited forever on a sentinel nothing would write. + name: "no overlay built: no workload container is mistaken for it", + containers: []DockerRunStruct{appC, sidecar}, + overlayName: "", + wantOK: false, + wantWorkload: []string{appC.Name, sidecar.Name}, + }, + { + name: "overlay name set but absent from the list", + containers: []DockerRunStruct{appC, sidecar}, + overlayName: overlay.Name, + wantOK: false, + wantWorkload: []string{appC.Name, sidecar.Name}, + }, + { + // Used to panic on containers[0]. + name: "empty container list", + containers: nil, + overlayName: overlay.Name, + wantOK: false, + wantWorkload: []string{}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + gotOverlay, gotWorkload, gotOK := selectNetworkOverlay(tc.containers, tc.overlayName) + + if gotOK != tc.wantOK { + t.Fatalf("ok = %v, want %v", gotOK, tc.wantOK) + } + if gotOverlay.Name != tc.wantOverlay { + t.Errorf("overlay = %q, want %q", gotOverlay.Name, tc.wantOverlay) + } + if got := names(gotWorkload); !equal(got, tc.wantWorkload) { + t.Errorf("workload = %v, want %v", got, tc.wantWorkload) + } + }) + } +} + +// The original list must not be modified: it is still used by the non-mesh +// startup path. +func TestSelectNetworkOverlayDoesNotMutateInput(t *testing.T) { + containers := []DockerRunStruct{ + {Name: "ns-uid-network-overlay"}, + {Name: "ns-uid-app"}, + {Name: "ns-uid-sidecar"}, + } + before := names(containers) + + if _, _, ok := selectNetworkOverlay(containers, "ns-uid-network-overlay"); !ok { + t.Fatal("expected the overlay to be found") + } + + if after := names(containers); !equal(before, after) { + t.Errorf("input list was mutated: %v -> %v", before, after) + } +} diff --git a/pkg/docker/Delete.go b/pkg/docker/Delete.go index dae88b6..2493182 100644 --- a/pkg/docker/Delete.go +++ b/pkg/docker/Delete.go @@ -5,13 +5,13 @@ import ( "io" "net/http" "os" + "strconv" "strings" "time" exec "github.com/alexellis/go-execute/pkg/v1" "github.com/containerd/containerd/log" - commonIL "github.com/interlink-hq/interlink/pkg/interlink" - "github.com/intertwin-eu/interlink-docker-plugin/pkg/docker/dindmanager" + commonIL "github.com/intertwin-eu/interlink-docker-plugin/pkg/common" "go.opentelemetry.io/otel" "go.opentelemetry.io/otel/attribute" trace "go.opentelemetry.io/otel/trace" @@ -57,7 +57,21 @@ func (h *SidecarHandler) DeleteHandler(w http.ResponseWriter, r *http.Request) { for _, container := range pod.Spec.Containers { containerName := podNamespace + "-" + podUID + "-" + container.Name - h.GpuManager.Release(containerName) + // if the FPGA manager is nil we don't need to release the container + if h.FPGAManager != nil { + // release the container from the FPGA manager + err = h.FPGAManager.Release(containerName) + if err != nil { + log.G(h.Ctx).Error("\u274C [DELETE CALL] Error releasing container " + containerName) + } + } + // same for the GPU manager: it is nil on nodes started without GPU support + if h.GpuManager != nil { + err = h.GpuManager.Release(containerName) + if err != nil { + log.G(h.Ctx).Error("\u274C [DELETE CALL] Error releasing GPUs of container " + containerName) + } + } } log.G(h.Ctx).Debug("\u2705 [DELETE CALL] Deleting POD " + podUID + "_dind") @@ -71,49 +85,40 @@ func (h *SidecarHandler) DeleteHandler(w http.ResponseWriter, r *http.Request) { execReturn, _ = shell.Execute() execReturn.Stdout = strings.ReplaceAll(execReturn.Stdout, "\n", "") - if execReturn.Stderr != "" { - log.G(h.Ctx).Error("\u274C [DELETE CALL] Error deleting container " + podUID + "_dind") - statusCode = http.StatusInternalServerError + // go-execute reports a nil error for non-zero exits, so inspect ExitCode + // rather than Stderr (docker prints benign warnings to stderr on success). + // Removing a missing container is NOT treated as fatal: a pod can fail before + // its DIND is ever created, and delete must stay idempotent \u2014 returning 500 + // here would make interLink retry the delete forever. + if execReturn.ExitCode != 0 { + log.G(h.Ctx).Warning("\u26A0\uFE0F [DELETE CALL] docker rm -f " + podUID + "_dind exited " + + strconv.Itoa(execReturn.ExitCode) + ": " + strings.TrimSpace(execReturn.Stderr)) } else { log.G(h.Ctx).Info("\u2705 [DELETE CALL] Deleted container " + podUID + "_dind") } - dindSpec := dindmanager.DindSpecs{} - dindSpec, err = h.DindManager.GetDindFromPodUID(podUID) - - if err != nil { - log.G(h.Ctx).Error("\u274C [DELETE CALL] Error retrieving DindSpecs, maybe the Dind container has already been deleted") + dindSpec, lookupErr := h.DindManager.GetDindFromPodUID(podUID) + if lookupErr != nil { + // The in-memory entry may already be gone (a failed create cleaned it, or + // it never got a pod UID). We cannot name its network here, but the + // orphan-network reaper below reclaims any unattached _dind_network. + log.G(h.Ctx).Info("\u2139\uFE0F [DELETE CALL] No DIND entry for pod " + podUID + " (already removed?)") } else { - log.G(h.Ctx).Info("\u2705 [DELETE CALL] Retrieved DindSpecs: " + dindSpec.DindID + " " + dindSpec.PodUID + " " + dindSpec.DindNetworkID + " ") - - // log the retrieved dindSpec - log.G(h.Ctx).Info("\u2705 [DELETE CALL] Retrieved DindSpecs: " + dindSpec.DindID + " " + dindSpec.PodUID + " " + dindSpec.DindNetworkID + " ") + log.G(h.Ctx).Info("\u2705 [DELETE CALL] Retrieved DindSpecs: " + dindSpec.DindID + " " + dindSpec.PodUID + " " + dindSpec.DindNetworkID) - cmd = []string{"network", "rm", dindSpec.DindNetworkID} - shell = exec.ExecTask{ - Command: "docker", - Args: cmd, - Shell: true, - } - execReturn, _ = shell.Execute() - execReturn.Stdout = strings.ReplaceAll(execReturn.Stdout, "\n", "") - if execReturn.Stderr != "" { - log.G(h.Ctx).Error("\u274C [DELETE CALL] Error deleting network " + dindSpec.DindNetworkID) - } else { - log.G(h.Ctx).Info("\u2705 [DELETE CALL] Deleted network " + dindSpec.DindNetworkID) + // RemoveDindFromList removes the bridge network and returns the subnet to + // the pool. The DIND container was already force-removed above, so its + // network no longer has an attached endpoint and can now be deleted. + if remErr := h.DindManager.RemoveDindFromList(dindSpec.PodUID); remErr != nil { + log.G(h.Ctx).Error("\u274C [DELETE CALL] Error removing DIND from list: " + remErr.Error()) } - // set the dind available again - err = h.DindManager.RemoveDindFromList(dindSpec.PodUID) - if err != nil { - log.G(h.Ctx).Error("\u274C [DELETE CALL] Error setting DIND container available") - } - } - wd, err := os.Getwd() - if err != nil { - HandleErrorAndRemoveData(h, w, "Unable to get current working directory", err, "", "") - return } - podDirectoryPathToDelete := filepath.Join(wd, h.Config.DataRootFolder+"/"+podNamespace+"-"+podUID) + + // Best-effort: reclaim any orphan DIND networks left behind by a create that + // failed before its entry was fully registered. + h.DindManager.ReapOrphanNetworks() + + podDirectoryPathToDelete := filepath.Join(h.Config.DataRootFolder, podNamespace+"-"+podUID) log.G(h.Ctx).Info("\u2705 [DELETE CALL] Deleting directory " + podDirectoryPathToDelete) err = os.RemoveAll(podDirectoryPathToDelete) diff --git a/pkg/docker/Status.go b/pkg/docker/Status.go index ec29116..72f75cc 100644 --- a/pkg/docker/Status.go +++ b/pkg/docker/Status.go @@ -28,7 +28,9 @@ func (h *SidecarHandler) StatusHandler(w http.ResponseWriter, r *http.Request) { attribute.Int64("start.timestamp", start), )) - var resp []commonIL.PodStatus + // Initialised (not just declared) so a response in which every pod was skipped + // marshals to [] rather than null. + resp := []commonIL.PodStatus{} var req []*v1.Pod statusCode := http.StatusOK @@ -50,7 +52,7 @@ func (h *SidecarHandler) StatusHandler(w http.ResponseWriter, r *http.Request) { return } - for i, pod := range req { + for _, pod := range req { podUID := string(pod.UID) podNamespace := string(pod.Namespace) @@ -74,21 +76,51 @@ func (h *SidecarHandler) StatusHandler(w http.ResponseWriter, r *http.Request) { //dindUUID := strings.ReplaceAll(execReturn.Stdout, "\n", "") dindUUID := strings.Join(strings.Fields(execReturn.Stdout), "") - log.G(h.Ctx).Info("\u2705 [STATUS CALL] UUID of the dind container retrieved successfully: ", dindUUID) - // if the string is empty or the length of the string is 0, return an error and 404 status code\ - if len(dindUUID) == 0 || dindUUID == "" { - log.G(h.Ctx).Error("\u274C [STATUS CALL] Error retrieving UUID of the dind container") - statusCode = http.StatusNotFound - w.WriteHeader(statusCode) - w.Write([]byte("DIND container with UUID " + dindUUID + " not found. Maybe it was deleted or never existed.")) - return + // go-execute reports a nil error for a non-zero exit, so an unknown container + // shows up here as empty stdout. Skip that pod instead of failing the whole + // request: interLink asks for the status of every pod on the node in one call, + // and answering 404 for the batch left every other pod unreconciled. Reporting + // no status for this pod is also safer than inventing a terminal one \u2014 the DIND + // is legitimately absent for a short window between the claim and the rename. + if dindUUID == "" { + log.G(h.Ctx).Warning("\u26A0\uFE0F [STATUS CALL] No DIND container found for pod " + podUID + + " (not created yet, or already deleted); skipping it in this status response") + continue } - resp = append(resp, commonIL.PodStatus{PodName: pod.Name, PodUID: podUID, PodNamespace: podNamespace, JobID: dindUUID}) + log.G(h.Ctx).Info("\u2705 [STATUS CALL] UUID of the dind container retrieved successfully: ", dindUUID) + + podStatus := commonIL.PodStatus{PodName: pod.Name, PodUID: podUID, PodNamespace: podNamespace, JobID: dindUUID} + + disabledInitContainers := make(map[string]bool) + if ann, ok := pod.Annotations["interlink.eu/disable-offload-init-containers"]; ok { + for _, name := range strings.Split(ann, ",") { + name = strings.TrimSpace(name) + if name != "" { + disabledInitContainers[name] = true + } + } + } // check if the pod has initContainers and get their status for _, container := range pod.Spec.InitContainers { + + if disabledInitContainers[container.Name] { + log.G(h.Ctx).Infof("✅ [STATUS CALL] init container %s is marked as non-offloaded, reporting as Completed", container.Name) + podStatus.InitContainers = append(podStatus.InitContainers, v1.ContainerStatus{ + Name: container.Name, + Ready: false, + State: v1.ContainerState{ + Terminated: &v1.ContainerStateTerminated{ + ExitCode: 0, + Reason: "Completed", + }, + }, + }) + continue + } + containerName := podNamespace + "-" + podUID + "-" + container.Name cmd := []string{"exec " + podUID + "_dind" + " docker ps -af name=^" + containerName + "$ --format \"{{.Status}}\""} shell := exec.ExecTask{ @@ -113,9 +145,9 @@ func (h *SidecarHandler) StatusHandler(w http.ResponseWriter, r *http.Request) { log.G(h.Ctx).Info("\u2705 [STATUS CALL] The container " + container.Name + " is in the state: " + initContainerStatus[0]) if initContainerStatus[0] == "Created" { - resp[i].InitContainers = append(resp[i].InitContainers, v1.ContainerStatus{Name: container.Name, State: v1.ContainerState{Waiting: &v1.ContainerStateWaiting{}}, Ready: false}) + podStatus.InitContainers = append(podStatus.InitContainers, v1.ContainerStatus{Name: container.Name, State: v1.ContainerState{Waiting: &v1.ContainerStateWaiting{}}, Ready: false}) } else if initContainerStatus[0] == "Up" { - resp[i].InitContainers = append(resp[i].InitContainers, v1.ContainerStatus{Name: container.Name, State: v1.ContainerState{Running: &v1.ContainerStateRunning{}}, Ready: true}) + podStatus.InitContainers = append(podStatus.InitContainers, v1.ContainerStatus{Name: container.Name, State: v1.ContainerState{Running: &v1.ContainerStateRunning{}}, Ready: true}) } else if initContainerStatus[0] == "Exited" { containerExitCode := strings.Split(initContainerStatus[1], "(") exitCode, err := strconv.Atoi(strings.Trim(containerExitCode[1], ")")) @@ -123,15 +155,37 @@ func (h *SidecarHandler) StatusHandler(w http.ResponseWriter, r *http.Request) { log.G(h.Ctx).Error(err) exitCode = 0 } - resp[i].InitContainers = append(resp[i].InitContainers, v1.ContainerStatus{Name: container.Name, State: v1.ContainerState{Terminated: &v1.ContainerStateTerminated{ExitCode: int32(exitCode)}}, Ready: false}) + podStatus.InitContainers = append(podStatus.InitContainers, v1.ContainerStatus{Name: container.Name, State: v1.ContainerState{Terminated: &v1.ContainerStateTerminated{ExitCode: int32(exitCode)}}, Ready: false}) } } else { - resp[i].InitContainers = append(resp[i].InitContainers, v1.ContainerStatus{Name: container.Name, State: v1.ContainerState{Waiting: &v1.ContainerStateWaiting{}}, Ready: false}) + podStatus.InitContainers = append(podStatus.InitContainers, v1.ContainerStatus{Name: container.Name, State: v1.ContainerState{Waiting: &v1.ContainerStateWaiting{}}, Ready: false}) + } + } + + disabledContainers := make(map[string]bool) + if ann, ok := pod.Annotations["interlink.eu/disable-offload-containers"]; ok { + for _, name := range strings.Split(ann, ",") { + name = strings.TrimSpace(name) + if name != "" { + disabledContainers[name] = true + } } } for _, container := range pod.Spec.Containers { + if disabledContainers[container.Name] { + log.G(h.Ctx).Infof("✅ [STATUS CALL] container %s is marked as non-offloaded, reporting as Running", container.Name) + podStatus.Containers = append(podStatus.Containers, v1.ContainerStatus{ + Name: container.Name, + Ready: true, + State: v1.ContainerState{ + Running: &v1.ContainerStateRunning{}, + }, + }) + continue + } + containerName := podNamespace + "-" + podUID + "-" + container.Name cmd := []string{"exec " + podUID + "_dind" + " docker ps -af name=^" + containerName + "$ --format \"{{.Status}}\""} @@ -157,9 +211,9 @@ func (h *SidecarHandler) StatusHandler(w http.ResponseWriter, r *http.Request) { log.G(h.Ctx).Info("\u2705 [STATUS CALL] The container " + container.Name + " is in the state: " + containerstatus[0]) if containerstatus[0] == "Created" { - resp[i].Containers = append(resp[i].Containers, v1.ContainerStatus{Name: container.Name, State: v1.ContainerState{Waiting: &v1.ContainerStateWaiting{}}, Ready: false}) + podStatus.Containers = append(podStatus.Containers, v1.ContainerStatus{Name: container.Name, State: v1.ContainerState{Waiting: &v1.ContainerStateWaiting{}}, Ready: false}) } else if containerstatus[0] == "Up" { - resp[i].Containers = append(resp[i].Containers, v1.ContainerStatus{Name: container.Name, State: v1.ContainerState{Running: &v1.ContainerStateRunning{}}, Ready: true}) + podStatus.Containers = append(podStatus.Containers, v1.ContainerStatus{Name: container.Name, State: v1.ContainerState{Running: &v1.ContainerStateRunning{}}, Ready: true}) } else if containerstatus[0] == "Exited" { containerExitCode := strings.Split(containerstatus[1], "(") exitCode, err := strconv.Atoi(strings.Trim(containerExitCode[1], ")")) @@ -167,14 +221,14 @@ func (h *SidecarHandler) StatusHandler(w http.ResponseWriter, r *http.Request) { log.G(h.Ctx).Error(err) exitCode = 0 } - resp[i].Containers = append(resp[i].Containers, v1.ContainerStatus{Name: container.Name, State: v1.ContainerState{Terminated: &v1.ContainerStateTerminated{ExitCode: int32(exitCode)}}, Ready: false}) - // release all the GPUs from the container - //h.GpuManager.Release(containerName) + podStatus.Containers = append(podStatus.Containers, v1.ContainerStatus{Name: container.Name, State: v1.ContainerState{Terminated: &v1.ContainerStateTerminated{ExitCode: int32(exitCode)}}, Ready: false}) } } else { - resp[i].Containers = append(resp[i].Containers, v1.ContainerStatus{Name: container.Name, State: v1.ContainerState{Waiting: &v1.ContainerStateWaiting{}}, Ready: false}) + podStatus.Containers = append(podStatus.Containers, v1.ContainerStatus{Name: container.Name, State: v1.ContainerState{Waiting: &v1.ContainerStateWaiting{}}, Ready: false}) } } + + resp = append(resp, podStatus) } w.WriteHeader(statusCode) diff --git a/pkg/docker/dindmanager/DindHandler.go b/pkg/docker/dindmanager/DindHandler.go index 676d5fd..d4c2691 100644 --- a/pkg/docker/dindmanager/DindHandler.go +++ b/pkg/docker/dindmanager/DindHandler.go @@ -4,14 +4,17 @@ import ( "context" "crypto/rand" "fmt" + "net" "os" "strings" + "sync" "time" exec "github.com/alexellis/go-execute/pkg/v1" "github.com/containerd/containerd/log" OSexec "os/exec" + "sync/atomic" ) type DindManagerInterface interface { @@ -19,45 +22,207 @@ type DindManagerInterface interface { BuildDindContainers(nDindContainer int8) error PrintDindList() error GetAvailableDind() (string, error) + ClaimAvailableDind(podUID string) (string, error) SetDindUnavailable(dindID string) error RemoveDindFromList(PodUID string) error SetPodUIDToDind(dindID string, podUID string) error GetDindFromPodUID(podUID string) (DindSpecs, error) SetDindAvailable(PodUID string) error + ReapOrphanNetworks() } type DindSpecs struct { - DindID string - PodUID string - DindNetworkID string - Available bool + DindID string + PodUID string + DindNetworkID string + AllocatedSubnet string // empty if no subnet pool was configured + Available bool } type DindManager struct { - DindList []DindSpecs - Ctx context.Context + DindList []DindSpecs + Ctx context.Context + FPGAEnabled bool + XilinxToolsPath string + + // Subnet pool: pre-generated /24 subnets from the parent CIDRs in config. + // Protected by mu because container creation/deletion can happen concurrently. + SubnetPool []string + mu sync.Mutex + InitialPoolSz int // total subnets at startup, used for logging + + // listMu protects DindList. Kept separate from mu (which protects SubnetPool) + // so that RemoveDindFromList can hold listMu while calling freeSubnet (mu) + // without self-deadlock. Lock ordering is always listMu -> mu, never reverse. + listMu sync.Mutex + + // buildsInFlight counts in-progress BuildDindContainers calls. The orphan + // network reaper checks it so it never removes a bridge network in the brief + // window after it is created but before its DIND container attaches to it. + buildsInFlight int32 } -// GenerateUUIDv4 generates a random UUIDv4 +// --------------------------------------------------------------------------- +// Subnet pool helpers +// --------------------------------------------------------------------------- + +// InitSubnetPool expands a list of parent CIDRs (e.g. ["192.168.0.0/16", +// "10.12.0.0/16"]) into an ordered slice of allocatable /24 subnets. +// Supported parent prefix lengths: /8, /16, /24. +// Returns an empty slice (no error) when parentCIDRs is nil/empty. +func InitSubnetPool(parentCIDRs []string) ([]string, error) { + var pool []string + for _, cidr := range parentCIDRs { + subnets, err := generateSubnetsFromCIDR(cidr) + if err != nil { + return nil, fmt.Errorf("failed to expand CIDR %s: %w", cidr, err) + } + pool = append(pool, subnets...) + } + return pool, nil +} + +// generateSubnetsFromCIDR returns all usable /24 subnets within parentCIDR. +// +// - /8 → iterates second octet [1-254] × third octet [1-254] (~64 k subnets) +// - /16 → iterates third octet [1-254] (254 subnets) +// - /24 → returns the CIDR itself (1 subnet) +func generateSubnetsFromCIDR(cidr string) ([]string, error) { + _, ipNet, err := net.ParseCIDR(cidr) + if err != nil { + return nil, fmt.Errorf("invalid CIDR %q: %w", cidr, err) + } + + ones, bits := ipNet.Mask.Size() + if bits != 32 { + return nil, fmt.Errorf("only IPv4 CIDRs are supported (got %s)", cidr) + } + + base := ipNet.IP.To4() + var subnets []string + + switch ones { + case 8: + for second := 1; second <= 254; second++ { + for third := 1; third <= 254; third++ { + subnets = append(subnets, fmt.Sprintf("%d.%d.%d.0/24", base[0], second, third)) + } + } + case 16: + for third := 1; third <= 254; third++ { + subnets = append(subnets, fmt.Sprintf("%d.%d.%d.0/24", base[0], base[1], third)) + } + case 24: + subnets = append(subnets, cidr) + default: + return nil, fmt.Errorf("unsupported prefix length /%d in %s: only /8, /16 and /24 are supported", ones, cidr) + } + + return subnets, nil +} + +// allocateSubnet pops the next available subnet from the pool. +// Returns ("", false) when the pool is empty or was never configured. +func (a *DindManager) allocateSubnet() (string, bool) { + a.mu.Lock() + defer a.mu.Unlock() + if len(a.SubnetPool) == 0 { + return "", false + } + subnet := a.SubnetPool[0] + a.SubnetPool = a.SubnetPool[1:] + return subnet, true +} + +// freeSubnet returns a subnet to the pool (appended at the end so the pool +// stays ordered and previously-used ranges are re-used last). +func (a *DindManager) freeSubnet(subnet string) { + if subnet == "" { + return + } + a.mu.Lock() + defer a.mu.Unlock() + a.SubnetPool = append(a.SubnetPool, subnet) +} + +// remainingSubnets returns the current pool size (thread-safe). +func (a *DindManager) remainingSubnets() int { + a.mu.Lock() + defer a.mu.Unlock() + return len(a.SubnetPool) +} + +// rollbackDind undoes a partially created DIND: it force-removes the container, +// then its bridge network, and only then returns the subnet to the pool. +// +// The order matters. A running container holds an endpoint on its network, so +// Docker refuses to remove the network while it exists, and a network that still +// exists keeps owning its /24. Returning the subnet to the pool without tearing +// those down (as the failure paths used to do) hands out an address range Docker +// still considers allocated, so the next "docker network create --subnet" for it +// fails with "Pool overlaps with other one on this address space" — permanently, +// since ReapOrphanNetworks cannot remove a network whose container is alive. +func (a *DindManager) rollbackDind(randUID string, allocatedSubnet string) { + rm := exec.ExecTask{ + Command: "docker", + Args: []string{"rm", "-f", randUID + "_dind"}, + Shell: true, + } + if execReturn, err := rm.Execute(); err != nil || execReturn.ExitCode != 0 { + log.G(a.Ctx).Warn(fmt.Sprintf("⚠️ Rollback: could not remove container %s_dind: %s", + randUID, strings.TrimSpace(execReturn.Stderr))) + } + + rmNet := exec.ExecTask{ + Command: "docker", + Args: []string{"network", "rm", randUID + "_dind_network"}, + Shell: true, + } + execReturn, err := rmNet.Execute() + // "not found" means the network is already gone, so the range is free and the + // subnet can be recycled: docker network rm is not idempotent and reports it + // as a failure. + alreadyGone := strings.Contains(execReturn.Stderr, "not found") + if (err != nil || execReturn.ExitCode != 0) && !alreadyGone { + // The subnet is deliberately NOT returned to the pool in this case: the + // network still exists and still owns the range, so reusing it would fail. + log.G(a.Ctx).Error(fmt.Sprintf( + "❌ Rollback: could not remove network %s_dind_network (%s); subnet %s is leaked rather than returned to the pool to avoid an overlap on reuse", + randUID, strings.TrimSpace(execReturn.Stderr), allocatedSubnet)) + return + } + + a.freeSubnet(allocatedSubnet) + if allocatedSubnet != "" { + log.G(a.Ctx).Info(fmt.Sprintf( + "✅ Rollback: subnet %s returned to pool (%d/%d now available)", + allocatedSubnet, a.remainingSubnets(), a.InitialPoolSz)) + } +} + +// --------------------------------------------------------------------------- +// UUIDv4 generator (unchanged) +// --------------------------------------------------------------------------- + func GenerateUUIDv4() (string, error) { uuid := make([]byte, 16) _, err := rand.Read(uuid) if err != nil { return "", err } - uuid[6] = (uuid[6] & 0x0f) | 0x40 uuid[8] = (uuid[8] & 0x3f) | 0x80 - - return fmt.Sprintf("%08x-%04x-%04x-%04x-%012x", uuid[0:4], uuid[4:6], uuid[6:8], uuid[8:10], uuid[10:16]), nil + return fmt.Sprintf("%08x-%04x-%04x-%04x-%012x", + uuid[0:4], uuid[4:6], uuid[6:8], uuid[8:10], uuid[10:16]), nil } -func (a *DindManager) CleanDindContainers() error { +// --------------------------------------------------------------------------- +// DindManager methods +// --------------------------------------------------------------------------- - // print the number of DIND containers to be created - log.G(a.Ctx).Info(fmt.Sprintf("\u2705 Start cleaning zombie DIND containers")) +func (a *DindManager) CleanDindContainers() error { + log.G(a.Ctx).Info("\u2705 Start cleaning zombie DIND containers") - // exec this command docker ps -a --format "{{.Names}}" | grep '_dind$' | wc -l shell := exec.ExecTask{ Command: "docker", Args: []string{"ps", "-a", "--format", "{{.Names}}", "|", "grep", "_dind$", "|", "wc", "-l"}, @@ -67,257 +232,292 @@ func (a *DindManager) CleanDindContainers() error { if err != nil { return err } - - // log the number of zombie DIND containers (remove the \n at the end of the string) - log.G(a.Ctx).Info(fmt.Sprintf("\u2705 %s zombie DIND containers found", strings.ReplaceAll(execReturn.Stdout, "\n", ""))) + log.G(a.Ctx).Info(fmt.Sprintf("\u2705 %s zombie DIND containers found", + strings.ReplaceAll(execReturn.Stdout, "\n", ""))) shell = exec.ExecTask{ Command: "docker", Args: []string{"ps", "-a", "--format", "{{.Names}}", "|", "grep", "_dind$", "|", "xargs", "-I", "{}", "docker", "rm", "-f", "{}"}, Shell: true, } - _, err = shell.Execute() - if err != nil { + if _, err = shell.Execute(); err != nil { return err } - // exec this command docker network ls --filter name=_dind_network$ --format "{{.ID}}" | xargs -r docker network rm - shell = exec.ExecTask{ Command: "docker", Args: []string{"network", "ls", "--filter", "name=_dind_network$", "--format", "{{.ID}}", "|", "xargs", "-r", "docker", "network", "rm"}, Shell: true, } - _, err = shell.Execute() - if err != nil { + if _, err = shell.Execute(); err != nil { return err } - log.G(a.Ctx).Info(fmt.Sprintf("\u2705 DIND zombie containers cleaned")) - + log.G(a.Ctx).Info("\u2705 DIND zombie containers cleaned") return nil } func (a *DindManager) BuildDindContainers(nDindContainer int8) error { + // Signal to the reaper that a build is running: freshly created networks are + // briefly container-less and must not be reaped as "orphans". + atomic.AddInt32(&a.buildsInFlight, 1) + defer atomic.AddInt32(&a.buildsInFlight, -1) - // print the number of DIND containers to be created log.G(a.Ctx).Info(fmt.Sprintf("\u2705 Creating %d DIND containers", nDindContainer)) - // get the working dir + // Log subnet pool status before we start allocating. + if a.InitialPoolSz > 0 { + log.G(a.Ctx).Info(fmt.Sprintf( + "\u2705 Subnet pool: %d/%d subnets available before container creation", + a.remainingSubnets(), a.InitialPoolSz, + )) + } else { + log.G(a.Ctx).Info("\u2705 No subnet pool configured — networks will use Docker's automatic addressing") + } + wd, err := os.Getwd() if err != nil { return err } - // get the env variable GPUENABLED, if 1 then the DIND container will have GPU support, otherwise it will not - gpuEnabled := os.Getenv("GPUENABLED") - dindImage := "docker:dind" + dindImage := "docker:29.3.0-dind" if gpuEnabled == "1" { dindImage = "ghcr.io/extrality/nvidia-dind" } for i := int8(0); i < nDindContainer; i++ { - // generate a random UID for the DIND container randUID, err := GenerateUUIDv4() if err != nil { return err } - // create the networks + // ---------------------------------------------------------------- + // Allocate a /24 subnet for this container's bridge network (if a + // pool is configured). + // ---------------------------------------------------------------- + allocatedSubnet, hasSubnet := a.allocateSubnet() + + networkArgs := []string{"network", "create", "--driver", "bridge"} + if hasSubnet { + networkArgs = append(networkArgs, "--subnet", allocatedSubnet) + } + networkArgs = append(networkArgs, randUID+"_dind_network") + shell := exec.ExecTask{ Command: "docker", - Args: []string{"network", "create", "--driver", "bridge", randUID + "_dind_network"}, + Args: networkArgs, Shell: true, } - _, err = shell.Execute() - - log.G(a.Ctx).Info(fmt.Sprintf("\u2705 DIND network %s created", randUID+"_dind_network")) - + // go-execute returns a nil error for a non-zero exit, so the exit code has to + // be checked explicitly. Otherwise a rejected "network create" (typically + // "Pool overlaps with other one on this address space") is treated as a + // success and the failure only surfaces later, as an unrelated "docker run" + // error, with the subnet already considered in use. + netReturn, err := shell.Execute() + if err == nil && netReturn.ExitCode != 0 { + err = fmt.Errorf("docker network create exited with code %d: %s", + netReturn.ExitCode, strings.TrimSpace(netReturn.Stderr)) + } if err != nil { + log.G(a.Ctx).Error(fmt.Sprintf("❌ Error creating DIND network %s_dind_network: %v", randUID, err)) + // The network was not created, so nothing to tear down: just return the + // subnet to the pool so it is not lost. + a.freeSubnet(allocatedSubnet) return err } + if hasSubnet { + log.G(a.Ctx).Info(fmt.Sprintf( + "\u2705 DIND network %s created with subnet %s (%d/%d subnets remaining in pool)", + randUID+"_dind_network", allocatedSubnet, + a.remainingSubnets(), a.InitialPoolSz, + )) + } else { + log.G(a.Ctx).Info(fmt.Sprintf( + "\u2705 DIND network %s created (no subnet pool configured)", + randUID+"_dind_network", + )) + } + + // ---------------------------------------------------------------- + // Build the docker-run argument list. + // ---------------------------------------------------------------- dindContainerArgs := []string{"run"} - //dindContainerArgs = append(dindContainerArgs, gpuArgsAsArray...) + if _, err := os.Stat("/cvmfs"); err == nil { dindContainerArgs = append(dindContainerArgs, "-v", "/cvmfs:/cvmfs") } - if os.Getenv("FPGAENABLED") == "1" { - if _, err := os.Stat("/tools/Xilinx/"); err == nil { - dindContainerArgs = append(dindContainerArgs, "-v", "/tools/Xilinx/:/tools/Xilinx/:ro") + if a.FPGAEnabled { + if _, err := os.Stat(a.XilinxToolsPath); err == nil { + dindContainerArgs = append(dindContainerArgs, "-v", a.XilinxToolsPath+":"+a.XilinxToolsPath+":ro") } } - // add the network to the dind container dindContainerArgs = append(dindContainerArgs, "--network", randUID+"_dind_network") - - // append also --net vk0 - //dindContainerArgs = append(dindContainerArgs, "--net", "vk0") dindContainerArgs = append(dindContainerArgs, "--cap-add", "NET_ADMIN") - // "--runtime=nvidia" is added to the dind container if the GPUENABLED env variable is set to 1 if gpuEnabled == "1" { dindContainerArgs = append(dindContainerArgs, "--runtime=nvidia") } - dindContainerArgs = append(dindContainerArgs, "--privileged", "-v", wd+":/"+wd, "-v", "/home:/home", "-v", "/var/lib/docker/overlay2:/var/lib/docker/overlay2", "-v", "/var/lib/docker/image:/var/lib/docker/image", "-d", "--name", randUID+"_dind", dindImage) + dindContainerArgs = append(dindContainerArgs, + "--privileged", + "-v", wd+":"+"/"+wd, + "-v", "/home:/home", + "-v", "/var/lib/docker/overlay2:/var/lib/docker/overlay2", + "-v", "/var/lib/docker/image:/var/lib/docker/image", + "-d", "--name", randUID+"_dind", dindImage, + ) - var dindContainerID string shell = exec.ExecTask{ Command: "docker", Args: dindContainerArgs, Shell: true, } - - // log the command to be executed with docker and the args log.G(a.Ctx).Info(fmt.Sprintf("\u2705 Command is %s", shell.Command+" "+strings.Join(shell.Args, " "))) execReturn, err := shell.Execute() + // go-execute returns a nil error for a non-zero exit, so the exit code has + // to be checked explicitly or a failed "docker run" is silently treated as + // a success (empty container ID, then a readiness timeout 20s later). + if err == nil && execReturn.ExitCode != 0 { + err = fmt.Errorf("docker run exited with code %d: %s", + execReturn.ExitCode, strings.TrimSpace(execReturn.Stderr)) + } if err != nil { - log.G(a.Ctx).Error(fmt.Sprintf("\u274c Error creating DIND container %s", randUID+"_dind")) + log.G(a.Ctx).Error(fmt.Sprintf("\u274c Error creating DIND container %s: %v", randUID+"_dind", err)) log.G(a.Ctx).Error(fmt.Sprintf("\u274c %s", execReturn.Stderr)) + // Tear down the network (and any half-started container) before the subnet + // goes back into the pool. + a.rollbackDind(randUID, allocatedSubnet) return err } - dindContainerID = execReturn.Stdout + dindContainerID := strings.TrimSpace(execReturn.Stdout) + if dindContainerID == "" { + a.rollbackDind(randUID, allocatedSubnet) + return fmt.Errorf("docker run returned an empty container ID for %s (stderr: %s)", + randUID+"_dind", strings.TrimSpace(execReturn.Stderr)) + } - // create a variable of maximum number of retries + // ---------------------------------------------------------------- + // Wait for the daemon inside the DinD container to be ready. + // ---------------------------------------------------------------- maxRetries := 20 - output := []byte{} - - // wait until the dind container is up and running by check that the command docker ps inside of it does not return an error + lastOutput := "" for { - if maxRetries == 0 { - return fmt.Errorf("DIND container %s not up and running", dindContainerID) + // Surface what the daemon actually logged, otherwise the timeout + // is indistinguishable from a container that never started. + log.G(a.Ctx).Error(fmt.Sprintf("❌ Last output of \"docker logs %s\": %s", + randUID+"_dind", strings.TrimSpace(lastOutput))) + a.rollbackDind(randUID, allocatedSubnet) + return fmt.Errorf("DIND container %s did not become ready in time", dindContainerID) } cmd := OSexec.Command("docker", "logs", randUID+"_dind") - output, err = cmd.CombinedOutput() - - if err != nil { - time.Sleep(1 * time.Second) - } - - if strings.Contains(string(output), "API listen on /var/run/docker.sock") { + output, err := cmd.CombinedOutput() + lastOutput = string(output) + if err == nil && strings.Contains(string(output), "API listen on /var/run/docker.sock") { break - } else { - time.Sleep(1 * time.Second) } - - maxRetries -= 1 - + time.Sleep(1 * time.Second) + maxRetries-- } - log.G(a.Ctx).Info(fmt.Sprintf("\u2705 DIND container %s is up and running", dindContainerID)) - shell = exec.ExecTask{ - Command: "docker", - Args: []string{"exec", randUID + "_dind", "apt-get", "update"}, - Shell: true, - } - _, err = shell.Execute() - if err != nil { - return err - } - - log.G(a.Ctx).Info(fmt.Sprintf("\u2705 Executed apt-get update")) - - shell = exec.ExecTask{ - Command: "docker", - Args: []string{"exec", randUID + "_dind", "apt-get", "install", "-y", "net-tools"}, - Shell: true, - } - _, err = shell.Execute() - if err != nil { - return err - } - - log.G(a.Ctx).Info(fmt.Sprintf("\u2705 Installed net-tools")) - - shell = exec.ExecTask{ - Command: "docker", - Args: []string{"exec", randUID + "_dind", "apt-get", "install", "-y", "iproute2"}, - Shell: true, - } - _, err = shell.Execute() - if err != nil { - return err + // ---------------------------------------------------------------- + // Install required tools inside the DinD container. + // ---------------------------------------------------------------- + for _, pkg := range []string{"net-tools", "iproute2"} { + shell = exec.ExecTask{ + Command: "docker", + Args: []string{"exec", randUID + "_dind", "apt-get", "install", "-y", pkg}, + Shell: true, + } + if _, err = shell.Execute(); err != nil { + a.rollbackDind(randUID, allocatedSubnet) + return err + } + log.G(a.Ctx).Info(fmt.Sprintf("\u2705 Installed %s", pkg)) } - // log.G(a.Ctx).Info(fmt.Sprintf("\u2705 Installed iproute2")) - - // // run docker network connet vk0 container-id - // shell = exec.ExecTask{ - // Command: "docker", - // Args: []string{"network", "connect", "vk0", randUID + "_dind"}, - // Shell: true, - // } - // _, err = shell.Execute() - // if err != nil { - // return err - // } - - // shell = exec.ExecTask{ - // Command: "docker", - // Args: []string{"exec", randUID + "_dind", "docker", "network", "create", "--subnet", "10.244.12.0/24", "-o", "com.docker.network.bridge.name=vk0", "vk0", "-o", "com.docker.network.driver.mtu=1500"}, - // Shell: true, - // } - // _, err = shell.Execute() - // if err != nil { - // return err - // } - - // log.G(a.Ctx).Info(fmt.Sprintf("\u2705 Created network vk0")) - - // shell = exec.ExecTask{ - // Command: "docker", - // Args: []string{"exec", randUID + "_dind", "ip", "link", "set", "to-cluster", "master", "vk0"}, - // Shell: true, - // } - // _, err = shell.Execute() - // if err != nil { - // return err - // } - - // log.G(a.Ctx).Info(fmt.Sprintf("\u2705 Set vk0 as master")) - - // add the dind container to the list of DIND containers - a.DindList = append(a.DindList, DindSpecs{DindID: randUID + "_dind", PodUID: "", DindNetworkID: randUID + "_dind_network", Available: true}) + // ---------------------------------------------------------------- + // Register the new DinD in the list. + // ---------------------------------------------------------------- + a.listMu.Lock() + a.DindList = append(a.DindList, DindSpecs{ + DindID: randUID + "_dind", + PodUID: "", + DindNetworkID: randUID + "_dind_network", + AllocatedSubnet: allocatedSubnet, // empty string when no pool + Available: true, + }) + a.listMu.Unlock() } return nil } func (a *DindManager) PrintDindList() error { - for _, dindSpec := range a.DindList { - log.G(a.Ctx).Info(fmt.Sprintf("DindID: %s, PodUID: %s, DindNetworkID: %s, Available: %t", dindSpec.DindID, dindSpec.PodUID, dindSpec.DindNetworkID, dindSpec.Available)) + a.listMu.Lock() + defer a.listMu.Unlock() + for _, d := range a.DindList { + log.G(a.Ctx).Info(fmt.Sprintf( + "DindID: %s, PodUID: %s, DindNetworkID: %s, AllocatedSubnet: %q, Available: %t", + d.DindID, d.PodUID, d.DindNetworkID, d.AllocatedSubnet, d.Available, + )) } return nil } func (a *DindManager) GetDindFromPodUID(podUID string) (DindSpecs, error) { - for _, dindSpec := range a.DindList { - if dindSpec.PodUID == podUID { - return dindSpec, nil + a.listMu.Lock() + defer a.listMu.Unlock() + for _, d := range a.DindList { + if d.PodUID == podUID { + return d, nil } } return DindSpecs{}, fmt.Errorf("DIND container with PodUID %s not found", podUID) } func (a *DindManager) GetAvailableDind() (string, error) { - for _, dindSpec := range a.DindList { - if dindSpec.Available { - return dindSpec.DindID, nil + a.listMu.Lock() + defer a.listMu.Unlock() + for _, d := range a.DindList { + if d.Available { + return d.DindID, nil + } + } + return "", fmt.Errorf("no available DIND container") +} + +// ClaimAvailableDind atomically finds an available DIND container, marks it +// unavailable, and records the pod UID against it — all under a single lock. +// Doing this in one step (instead of GetAvailableDind + SetDindUnavailable + +// SetPodUIDToDind) prevents two concurrent creates from grabbing the same +// container, and guarantees the DIND is findable by pod UID for cleanup the +// moment it is claimed, closing the window where a failure would orphan it. +func (a *DindManager) ClaimAvailableDind(podUID string) (string, error) { + a.listMu.Lock() + defer a.listMu.Unlock() + for i, d := range a.DindList { + if d.Available { + a.DindList[i].Available = false + a.DindList[i].PodUID = podUID + return a.DindList[i].DindID, nil } } - return "", fmt.Errorf("No available DIND container") + return "", fmt.Errorf("no available DIND container") } func (a *DindManager) SetDindUnavailable(dindID string) error { - for i, dindSpec := range a.DindList { - if dindSpec.DindID == dindID { + a.listMu.Lock() + defer a.listMu.Unlock() + for i, d := range a.DindList { + if d.DindID == dindID { a.DindList[i].Available = false return nil } @@ -325,19 +525,23 @@ func (a *DindManager) SetDindUnavailable(dindID string) error { return fmt.Errorf("DIND container %s not found", dindID) } -func (a *DindManager) SetDindAvailable(PodUI string) error { - for i, dindSpec := range a.DindList { - if dindSpec.PodUID == PodUI { +func (a *DindManager) SetDindAvailable(PodUID string) error { + a.listMu.Lock() + defer a.listMu.Unlock() + for i, d := range a.DindList { + if d.PodUID == PodUID { a.DindList[i].Available = true return nil } } - return fmt.Errorf("DIND container %s not found", PodUI) + return fmt.Errorf("DIND container with PodUID %s not found", PodUID) } func (a *DindManager) SetPodUIDToDind(dindID string, podUID string) error { - for i, dindSpec := range a.DindList { - if dindSpec.DindID == dindID { + a.listMu.Lock() + defer a.listMu.Unlock() + for i, d := range a.DindList { + if d.DindID == dindID { a.DindList[i].PodUID = podUID return nil } @@ -345,12 +549,82 @@ func (a *DindManager) SetPodUIDToDind(dindID string, podUID string) error { return fmt.Errorf("DIND container %s not found", dindID) } +// ReapOrphanNetworks removes DIND bridge networks that have no attached +// container. A create that fails after its network is created leaves such a +// network behind; because it still holds its /24 subnet, a later create drawing +// the same subnet from the pool fails with an address-overlap error. Removing +// only unattached networks is safe: Docker refuses to remove a network still in +// use by a live DIND container, so those are skipped. The build-in-flight guard +// avoids racing a network that was just created but whose container has not yet +// attached. +func (a *DindManager) ReapOrphanNetworks() { + if atomic.LoadInt32(&a.buildsInFlight) > 0 { + return + } + list := exec.ExecTask{ + Command: "docker", + Args: []string{"network", "ls", "--filter", "name=_dind_network", "--format", "{{.Name}}"}, + Shell: true, + } + execReturn, err := list.Execute() + if err != nil || execReturn.ExitCode != 0 { + return + } + for _, name := range strings.Fields(execReturn.Stdout) { + rm := exec.ExecTask{ + Command: "docker", + Args: []string{"network", "rm", name}, + Shell: true, + } + // Fails harmlessly (non-zero exit) if the network still has a live + // container attached; only truly orphaned networks are removed. + if r, _ := rm.Execute(); r.ExitCode == 0 { + log.G(a.Ctx).Info("🧹 Reaped orphan DIND network " + name) + } + } +} + +// RemoveDindFromList removes the DinD entry associated with PodUID from the +// in-memory list, tears down its Docker network, and returns its subnet to +// the pool so it can be reused by future containers. func (a *DindManager) RemoveDindFromList(PodUID string) error { - for i, dindSpec := range a.DindList { - if dindSpec.PodUID == PodUID { - a.DindList = append(a.DindList[:i], a.DindList[i+1:]...) - return nil + a.listMu.Lock() + defer a.listMu.Unlock() + for i, d := range a.DindList { + if d.PodUID != PodUID { + continue } + + // Remove the dedicated bridge network from Docker. + if d.DindNetworkID != "" { + shell := exec.ExecTask{ + Command: "docker", + Args: []string{"network", "rm", d.DindNetworkID}, + Shell: true, + } + if execReturn, err := shell.Execute(); err != nil { + // Log but don't abort — the container may already be gone. + log.G(a.Ctx).Warn(fmt.Sprintf( + "\u26a0\ufe0f Could not remove network %s: %s (stderr: %s)", + d.DindNetworkID, err, execReturn.Stderr, + )) + } else { + log.G(a.Ctx).Info(fmt.Sprintf("\u2705 Removed Docker network %s", d.DindNetworkID)) + } + } + + // Return the subnet to the pool. + a.freeSubnet(d.AllocatedSubnet) + if d.AllocatedSubnet != "" { + log.G(a.Ctx).Info(fmt.Sprintf( + "\u2705 Subnet %s returned to pool (%d/%d now available)", + d.AllocatedSubnet, a.remainingSubnets(), a.InitialPoolSz, + )) + } + + // Splice the entry out of the list. + a.DindList = append(a.DindList[:i], a.DindList[i+1:]...) + return nil } return fmt.Errorf("DIND container with PodUID %s not found", PodUID) } diff --git a/pkg/docker/fpgastrategies/AMDHandler.go b/pkg/docker/fpgastrategies/AMDHandler.go index bbb65cb..cd510d1 100644 --- a/pkg/docker/fpgastrategies/AMDHandler.go +++ b/pkg/docker/fpgastrategies/AMDHandler.go @@ -18,11 +18,14 @@ import ( ) type FPGASpecs struct { - BDF string - Shell string - LogicUUID string - deviceID string - ContainerID string + BDF string + Shell string + LogicUUID string + deviceID string + // ContainerIDs holds every container currently assigned this FPGA. It is + // normally at most one entry; it can hold more than one only when + // FPGA_DISABLE_BOOKKEEPING=1 lets containers share a device. + ContainerIDs []string DeviceReady string DeviceToMount string Available bool @@ -34,6 +37,8 @@ type FPGAManager struct { FPGASpecsMutex sync.Mutex Vendor string Ctx context.Context + VitisPath string + XRTPath string } type FPGAManagerInterface interface { @@ -44,7 +49,7 @@ type FPGAManagerInterface interface { Discover() error Check() error GetAvailableFPGAs(numFPGAs int) ([]FPGASpecs, error) - Assign(UUID string, containerID string) error + Assign(BDF string, containerID string) error Release(UUID string) error GetAndAssignAvailableFPGAs(numFPGAs int, containerID string) ([]FPGASpecs, error) } @@ -52,17 +57,17 @@ type FPGAManagerInterface interface { func (a *FPGAManager) Init() error { // Check if the Xilinx setup.sh file exists - if _, err := os.Stat("/opt/xilinx/xrt/setup.sh"); os.IsNotExist(err) { - return fmt.Errorf("/opt/xilinx/xrt/setup.sh does not exist: %v", err) + if _, err := os.Stat(a.XRTPath + "/setup.sh"); os.IsNotExist(err) { + return fmt.Errorf("%s/setup.sh does not exist: %v", a.XRTPath, err) } // Check if the path to Vitis exists - if _, err := os.Stat("/tools/Xilinx/Vitis/2023.2"); os.IsNotExist(err) { - return fmt.Errorf("/tools/Xilinx/Vitis/2023.2 does not exist: %v", err) + if _, err := os.Stat(a.VitisPath); os.IsNotExist(err) { + return fmt.Errorf("%s does not exist: %v", a.VitisPath, err) } // Source the setup.sh to initialize Xilinx tools using shell - shellArgs := []string{"source", "/opt/xilinx/xrt/setup.sh"} + shellArgs := []string{"source", a.XRTPath + "/setup.sh"} shell := exec.ExecTask{ Command: "/bin/bash", Args: shellArgs, @@ -88,103 +93,136 @@ func (a *FPGAManager) Discover() error { Shell: true, } - regexPattern := `^\[[0-9a-fA-F]{4}:[0-9a-fA-F]{2}:[0-9a-fA-F]{2}\.[0-9a-fA-F]\].*` - re := regexp.MustCompile(regexPattern) - - shell.Execute() output, err := shell.Execute() if err != nil { return fmt.Errorf("Error running lspci command: %v", err) } lines := strings.Split(string(output.Stdout), "\n") + xilinxFound := false for _, line := range lines { if strings.Contains(line, "Xilinx") { - parts := strings.Fields(line) - if len(parts) < 3 { - continue - } - bdf := parts[0] - shellArgs := []string{"/opt/xilinx/xrt/setup.sh"} - shell := exec.ExecTask{ - Command: "source", - Args: shellArgs, - Shell: true, - } + xilinxFound = true + log.G(a.Ctx).Info(fmt.Sprintf("\u2705 Found potential FPGA: %s", line)) + break + } + } - _, err := shell.Execute() - if err != nil { - return fmt.Errorf("Error running source setup.sh command: %v", err) - } + if !xilinxFound { + log.G(a.Ctx).Info("\u2705 No FPGAs discovered") + return nil + } - cmd := exec.ExecTask{ - Command: "/opt/xilinx/xrt/bin/xbutil", - Args: []string{"examine"}, // "--device", bdf - Shell: false, - } - outputXbutil, err := cmd.Execute() - if err != nil { - return fmt.Errorf("Error running xbutil examine: %v", err) - } + // Source XRT setup.sh before running xbutil + sourceShell := exec.ExecTask{ + Command: "source", + Args: []string{a.XRTPath + "/setup.sh"}, + Shell: true, + } + _, err = sourceShell.Execute() + if err != nil { + return fmt.Errorf("Error running source setup.sh command: %v", err) + } + + // Get a temp path but don't create the file — xbutil will create it + tmpFile, err := os.CreateTemp("", "xbutil-examine-*.json") + if err != nil { + return fmt.Errorf("Error creating temp file for xbutil output: %v", err) + } + tmpFilePath := tmpFile.Name() + tmpFile.Close() + os.Remove(tmpFilePath) // Remove it so xbutil can write to it freely + defer os.Remove(tmpFilePath) + + log.G(a.Ctx).Info(fmt.Sprintf("\u2705 Running xbutil examine -f json -o %s", tmpFilePath)) + + cmd := exec.ExecTask{ + Command: a.XRTPath + "/bin/xbutil", + Args: []string{"examine", "-f", "json", "-o", tmpFilePath}, + Shell: false, + } + result, err := cmd.Execute() + if err != nil { + return fmt.Errorf("Error running xbutil examine: %v", err) + } + + // Log stderr to help debug if JSON is still empty + if result.Stderr != "" { + log.G(a.Ctx).Info(fmt.Sprintf("xbutil stderr: %s", result.Stderr)) + } - examineLines := strings.Split(string(outputXbutil.Stdout), "\n") - for _, examineLine := range examineLines { - - if re.MatchString(examineLine) { - fpgas := strings.Split(examineLine, " : ") - fmt.Printf("FPGAs: %v\n", fpgas) - found := false - - deviceID := "" - reDeviceID := regexp.MustCompile(`user\(inst=(\d+)\)`) - matches := reDeviceID.FindStringSubmatch(fpgas[1]) - if len(matches) > 1 { - deviceID = matches[1] - } - - logicUUID := "" - reLogicUUID := regexp.MustCompile(`[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}`) - matches = reLogicUUID.FindStringSubmatch(fpgas[1]) - if len(matches) > 0 { - logicUUID = matches[0] - } - - shell := "" - reLogiShell := regexp.MustCompile(`xilinx_.*_base_`) - matches = reLogiShell.FindStringSubmatch(fpgas[1]) - if len(matches) > 0 { - shell = matches[0] - } - - for _, fpgaSpec := range a.FPGASpecsList { - if fpgaSpec.LogicUUID == logicUUID { - found = true - break - } - } - if len(fpgas) > 1 && !found { - spec := FPGASpecs{ - BDF: bdf, - Shell: shell, - LogicUUID: logicUUID, - deviceID: deviceID, - DeviceToMount: "/dev/dri/renderD" + deviceID, - Available: true, // Assuming it's available for now, can update based on further output parsing - } - a.FPGASpecsList = append(a.FPGASpecsList, spec) - } - } + // Verify the file was actually written + info, err := os.Stat(tmpFilePath) + if err != nil || info.Size() == 0 { + return fmt.Errorf("xbutil did not write output to %s (stderr: %s)", tmpFilePath, result.Stderr) + } + + // Parse the JSON output + jsonData, err := os.ReadFile(tmpFilePath) + if err != nil { + return fmt.Errorf("Error reading xbutil JSON output: %v", err) + } + + var examineOutput struct { + System struct { + Host struct { + Devices []struct { + BDF string `json:"bdf"` + VBNV string `json:"vbnv"` + ID string `json:"id"` + Instance string `json:"instance"` + IsReady string `json:"is_ready"` + } `json:"devices"` + } `json:"host"` + } `json:"system"` + } + + if err := json.Unmarshal(jsonData, &examineOutput); err != nil { + return fmt.Errorf("Error parsing xbutil JSON output: %v", err) + } + + reDeviceID := regexp.MustCompile(`user\(inst=(\d+)\)`) + + for _, device := range examineOutput.System.Host.Devices { + log.G(a.Ctx).Info(fmt.Sprintf("\u2705 Processing device BDF: %s, VBNV: %s, ID: %s", device.BDF, device.VBNV, device.ID)) + + // Check if already in the list + alreadyFound := false + for _, fpgaSpec := range a.FPGASpecsList { + if fpgaSpec.BDF == device.BDF { + alreadyFound = true + break } } + if alreadyFound { + continue + } + + deviceID := "" + if matches := reDeviceID.FindStringSubmatch(device.Instance); len(matches) > 1 { + deviceID = matches[1] + } + + spec := FPGASpecs{ + BDF: device.BDF, + Shell: device.VBNV, + LogicUUID: device.ID, + deviceID: deviceID, + DeviceToMount: "/dev/dri/renderD" + deviceID, + Available: device.IsReady == "true", + } + a.FPGASpecsList = append(a.FPGASpecsList, spec) } if len(a.FPGASpecsList) > 0 { log.G(a.Ctx).Info("\u2705 Discovered FPGAs:") for _, fpgaSpec := range a.FPGASpecsList { - log.G(a.Ctx).Info(fmt.Sprintf("\u2705 BDF: %s, Shell: %s, LogicUUID: %s, DeviceID %s, Available: %t", fpgaSpec.BDF, fpgaSpec.Shell, fpgaSpec.LogicUUID, fpgaSpec.deviceID, fpgaSpec.Available)) + log.G(a.Ctx).Info(fmt.Sprintf("\u2705 BDF: %s, Shell: %s, LogicUUID: %s, DeviceID: %s, Available: %t", + fpgaSpec.BDF, fpgaSpec.Shell, fpgaSpec.LogicUUID, fpgaSpec.deviceID, fpgaSpec.Available)) } + log.G(a.Ctx).Info(fmt.Sprintf("\u2705 Total FPGAs discovered: %d", len(a.FPGASpecsList))) } else { - log.G(a.Ctx).Info(" \u2705 No FPGAs discovered") + log.G(a.Ctx).Info("\u2705 No FPGAs discovered") } return nil @@ -203,30 +241,35 @@ func (a *FPGAManager) GetFPGASpecsList() []FPGASpecs { return a.FPGASpecsList } -func (a *FPGAManager) Assign(UUID string, containerID string) error { +// Assign locks FPGASpecsMutex itself, so it must not be called by a method +// that already holds it — use assignLocked in that case. +func (a *FPGAManager) Assign(BDF string, containerID string) error { + a.FPGASpecsMutex.Lock() + defer a.FPGASpecsMutex.Unlock() + return a.assignLocked(BDF, containerID) +} +// assignLocked adds containerID to the FPGA identified by BDF. Caller must +// hold FPGASpecsMutex. +func (a *FPGAManager) assignLocked(BDF string, containerID string) error { for i := range a.FPGASpecsList { - if a.FPGASpecsList[i].LogicUUID == UUID { - - // check if the BOOKKEEPING is disabled - disableBookkeeping := os.Getenv("FPGA_DISABLE_BOOKKEEPING") == "1" - if disableBookkeeping { - a.FPGASpecsList[i].ContainerID = containerID - a.FPGASpecsList[i].Available = false - break - } - - if !a.FPGASpecsList[i].Available { - return fmt.Errorf("FPGA with UUID %s is already in use by container %s", UUID, a.FPGASpecsList[i].ContainerID) - } + if a.FPGASpecsList[i].BDF != BDF { + continue + } - a.FPGASpecsList[i].ContainerID = containerID - a.FPGASpecsList[i].Available = false - break + disableBookkeeping := os.Getenv("FPGA_DISABLE_BOOKKEEPING") == "1" + if !disableBookkeeping && len(a.FPGASpecsList[i].ContainerIDs) > 0 { + return fmt.Errorf("FPGA with BDF %s is already in use by container(s) %v", BDF, a.FPGASpecsList[i].ContainerIDs) } - } - return nil + // Appending (instead of overwriting) is what makes sharing under + // FPGA_DISABLE_BOOKKEEPING actually work: every assignee stays tracked, + // so Release can later find each one instead of only the last assignee. + a.FPGASpecsList[i].ContainerIDs = append(a.FPGASpecsList[i].ContainerIDs, containerID) + a.FPGASpecsList[i].Available = false + return nil + } + return fmt.Errorf("FPGA with BDF %s not found", BDF) } func (a *FPGAManager) Release(containerID string) error { @@ -235,148 +278,105 @@ func (a *FPGAManager) Release(containerID string) error { defer a.FPGASpecsMutex.Unlock() for i := range a.FPGASpecsList { - if a.FPGASpecsList[i].ContainerID == containerID { - - if a.FPGASpecsList[i].Available { + ids := a.FPGASpecsList[i].ContainerIDs + for j, id := range ids { + if id != containerID { continue } - - a.FPGASpecsList[i].ContainerID = "" - a.FPGASpecsList[i].Available = true + a.FPGASpecsList[i].ContainerIDs = append(ids[:j], ids[j+1:]...) + if len(a.FPGASpecsList[i].ContainerIDs) == 0 { + a.FPGASpecsList[i].Available = true + } + break } } return nil } -/* func (a *FPGAManager) GetAvailableFPGAs(numFPGAs int) ([]FPGASpecs, error) { +// GetAvailableFPGAs locks FPGASpecsMutex itself, so it must not be called by +// a method that already holds it — use getAvailableFPGAsLocked in that case. +func (a *FPGAManager) GetAvailableFPGAs(numFPGAs int) ([]FPGASpecs, error) { a.FPGASpecsMutex.Lock() defer a.FPGASpecsMutex.Unlock() + return a.getAvailableFPGAsLocked(numFPGAs) +} +// getAvailableFPGAsLocked picks up to numFPGAs FPGAs. Caller must hold +// FPGASpecsMutex. +// +// Default behavior: only ever return fully-unassigned FPGAs. +// +// FPGA_DISABLE_BOOKKEEPING=1 relaxes that so callers can still get an answer +// once every device is in use: unassigned FPGAs are still preferred, but if +// there aren't enough, devices already in use are added back, least-shared +// first (fpgaUsage counts current assignees per BDF via len(ContainerIDs), so +// it reflects real concurrent sharing, not just "used at all"). +func (a *FPGAManager) getAvailableFPGAsLocked(numFPGAs int) ([]FPGASpecs, error) { var availableFPGAs []FPGASpecs - disableBookkeeping := os.Getenv("FPGA_DISABLE_BOOKKEEPING") == "1" + seen := make(map[string]bool) - if disableBookkeeping { - fpgaUsage := make(map[string]int) - for _, fpga := range a.FPGASpecsList { - if fpga.ContainerID != "" { - fpgaUsage[fpga.BDF]++ - } else { - // If an unassigned FPGA is found, prioritize it - availableFPGAs = append(availableFPGAs, fpga) - } - } - - if len(availableFPGAs) >= numFPGAs { - return availableFPGAs[:numFPGAs], nil - } - - // If not enough unassigned FPGAs, find the least assigned ones - sort.Slice(a.FPGASpecsList, func(i, j int) bool { - return fpgaUsage[a.FPGASpecsList[i].BDF] < fpgaUsage[a.FPGASpecsList[j].BDF] - }) - - for _, fpga := range a.FPGASpecsList { - if len(availableFPGAs) < numFPGAs { - availableFPGAs = append(availableFPGAs, fpga) - } else { - break - } - } + disableBookkeeping := os.Getenv("FPGA_DISABLE_BOOKKEEPING") == "1" - if len(availableFPGAs) >= numFPGAs { - return availableFPGAs[:numFPGAs], nil + for _, fpga := range a.FPGASpecsList { + if len(fpga.ContainerIDs) == 0 { + availableFPGAs = append(availableFPGAs, fpga) + seen[fpga.BDF] = true } - - return nil, fmt.Errorf("Not enough FPGAs available. Requested: %d, Found: %d", numFPGAs, len(availableFPGAs)) } - // Default behavior: return only available FPGAs - for _, fpgaSpec := range a.FPGASpecsList { - if fpgaSpec.Available { - availableFPGAs = append(availableFPGAs, fpgaSpec) - if len(availableFPGAs) == numFPGAs { - return availableFPGAs, nil - } - } + if len(availableFPGAs) >= numFPGAs { + return availableFPGAs[:numFPGAs], nil } - return nil, fmt.Errorf("Not enough available FPGAs. Requested: %d, Available: %d", numFPGAs, len(availableFPGAs)) -} */ - -func (a *FPGAManager) GetAvailableFPGAs(numFPGAs int) ([]FPGASpecs, error) { - - var availableFPGAs []FPGASpecs - - fmt.Println("Checking for available FPGAs") - - disableBookkeeping := os.Getenv("FPGA_DISABLE_BOOKKEEPING") == "1" - - fmt.Println(fmt.Sprintf("FPGA_DISABLE_BOOKKEEPING: %v", disableBookkeeping)) - - if disableBookkeeping { - fmt.Println("FPGA_DISABLE_BOOKKEEPING is set to 1. Disabling bookkeeping") - - fpgaUsage := make(map[string]int) - for _, fpga := range a.FPGASpecsList { - if fpga.ContainerID != "" { - fpgaUsage[fpga.BDF]++ - } else { - // If an unassigned FPGA is found, prioritize it - availableFPGAs = append(availableFPGAs, fpga) - } - } - - fmt.Println(fmt.Sprintf("FPGA Usage: %v", fpgaUsage)) + if !disableBookkeeping { + return nil, fmt.Errorf("Not enough available FPGAs. Requested: %d, Available: %d", numFPGAs, len(availableFPGAs)) + } - if len(availableFPGAs) >= numFPGAs { - return availableFPGAs[:numFPGAs], nil - } + log.G(a.Ctx).Info(fmt.Sprintf("✅ FPGA_DISABLE_BOOKKEEPING is set: only %d/%d unassigned FPGAs found, falling back to sharing least-used devices", len(availableFPGAs), numFPGAs)) - // If not enough unassigned FPGAs, find the least assigned ones - sort.Slice(a.FPGASpecsList, func(i, j int) bool { - return fpgaUsage[a.FPGASpecsList[i].BDF] < fpgaUsage[a.FPGASpecsList[j].BDF] - }) + fpgaUsage := make(map[string]int, len(a.FPGASpecsList)) + for _, fpga := range a.FPGASpecsList { + fpgaUsage[fpga.BDF] = len(fpga.ContainerIDs) + } - for _, fpga := range a.FPGASpecsList { - if len(availableFPGAs) < numFPGAs { - availableFPGAs = append(availableFPGAs, fpga) - } else { - break - } + shared := make([]FPGASpecs, len(a.FPGASpecsList)) + copy(shared, a.FPGASpecsList) + sort.Slice(shared, func(i, j int) bool { + return fpgaUsage[shared[i].BDF] < fpgaUsage[shared[j].BDF] + }) + + for _, fpga := range shared { + if seen[fpga.BDF] { + // Already included from the unassigned pass above — including it + // again would hand the caller the same physical device twice. + continue } - - if len(availableFPGAs) >= numFPGAs { - return availableFPGAs[:numFPGAs], nil + availableFPGAs = append(availableFPGAs, fpga) + seen[fpga.BDF] = true + if len(availableFPGAs) == numFPGAs { + break } - - return nil, fmt.Errorf("Not enough FPGAs available. Requested: %d, Found: %d", numFPGAs, len(availableFPGAs)) } - for _, fpgaSpec := range a.FPGASpecsList { - if fpgaSpec.Available { - availableFPGAs = append(availableFPGAs, fpgaSpec) - if len(availableFPGAs) == numFPGAs { - return availableFPGAs, nil - } - } + if len(availableFPGAs) >= numFPGAs { + return availableFPGAs[:numFPGAs], nil } - return nil, fmt.Errorf("Not enough available FPGAs. Requested: %d, Available: %d", numFPGAs, len(availableFPGAs)) + + return nil, fmt.Errorf("Not enough FPGAs available. Requested: %d, Found: %d", numFPGAs, len(availableFPGAs)) } func (a *FPGAManager) GetAndAssignAvailableFPGAs(numFPGAs int, containerID string) ([]FPGASpecs, error) { - a.FPGASpecsMutex.Lock() defer a.FPGASpecsMutex.Unlock() - fpgaSpecs, err := a.GetAvailableFPGAs(numFPGAs) + fpgaSpecs, err := a.getAvailableFPGAsLocked(numFPGAs) if err != nil { return nil, err } for _, fpgaSpec := range fpgaSpecs { - err = a.Assign(fpgaSpec.LogicUUID, containerID) - if err != nil { + if err := a.assignLocked(fpgaSpec.BDF, containerID); err != nil { return nil, err } } diff --git a/pkg/docker/func.go b/pkg/docker/func.go index 9752e3f..b5f031a 100644 --- a/pkg/docker/func.go +++ b/pkg/docker/func.go @@ -89,6 +89,16 @@ func NewInterLinkConfig() (DockerConfig, error) { InterLinkConfigInst.Tsockspath = path } + if InterLinkConfigInst.XilinxToolsPath == "" { + InterLinkConfigInst.XilinxToolsPath = "/tools/Xilinx/" + } + if InterLinkConfigInst.VitisPath == "" { + InterLinkConfigInst.VitisPath = "/tools/Xilinx/Vitis/2023.2" + } + if InterLinkConfigInst.XRTPath == "" { + InterLinkConfigInst.XRTPath = "/opt/xilinx/xrt" + } + InterLinkConfigInst.set = true } return InterLinkConfigInst, nil @@ -121,9 +131,9 @@ func SetDurationSpan(startTime int64, span trace.Span, opts ...SpanOption) { type SidecarHandler struct { Config DockerConfig Ctx context.Context - GpuManager gpustrategies.GPUManagerInterface DindManager dindmanager.DindManagerInterface FPGAManager fpgastrategies.FPGAManagerInterface + GpuManager gpustrategies.GPUManagerInterface FinalDNSNameserver string FinalDNSSearch string } @@ -146,18 +156,13 @@ func parseContainerCommandAndReturnArgs(Ctx context.Context, config DockerConfig prefileName := container.Name + "_" + podUID + "_" + podNamespace - wd, err := os.Getwd() - if err != nil { - return nil, nil, nil, err - } - if len(container.Command) > 0 { fileName := prefileName + "_script.sh" if len(container.Args) == 0 { - fileNamePath := filepath.Join(wd, config.DataRootFolder+podNamespace+"-"+podUID, fileName) - err = os.WriteFile(fileNamePath, []byte(strings.Join(container.Command, " ")), 0644) + fileNamePath := filepath.Join(config.DataRootFolder+podNamespace+"-"+podUID, fileName) + err := os.WriteFile(fileNamePath, []byte(strings.Join(container.Command, " ")), 0644) if err != nil { log.G(Ctx).Error(err) return nil, nil, nil, err @@ -166,15 +171,15 @@ func parseContainerCommandAndReturnArgs(Ctx context.Context, config DockerConfig } argsFileName := container.Name + "_args" - argsFileNamePath := filepath.Join(wd, config.DataRootFolder+podNamespace+"-"+podUID, argsFileName) - err = os.WriteFile(argsFileNamePath, []byte(strings.Join(container.Args, " ")), 0644) + argsFileNamePath := filepath.Join(config.DataRootFolder+podNamespace+"-"+podUID, argsFileName) + err := os.WriteFile(argsFileNamePath, []byte(strings.Join(container.Args, " ")), 0644) if err != nil { log.G(Ctx).Error(err) return nil, nil, nil, err } fullFileContent := strings.Join(container.Command, " ") + " \"$(cat " + argsFileName + ")\"" - fullFileNamePath := filepath.Join(wd, config.DataRootFolder+podNamespace+"-"+podUID, fileName) + fullFileNamePath := filepath.Join(config.DataRootFolder+podNamespace+"-"+podUID, fileName) err = os.WriteFile(fullFileNamePath, []byte(fullFileContent), 0644) if err != nil { log.G(Ctx).Error(err) @@ -254,12 +259,6 @@ func prepareMounts(Ctx context.Context, config DockerConfig, data commonIL.Retri } func mountData(Ctx context.Context, config DockerConfig, pod v1.Pod, data interface{}, container v1.Container) ([]string, error) { - wd, err := os.Getwd() - if err != nil { - log.G(Ctx).Error(err) - return nil, err - } - for _, mountSpec := range container.VolumeMounts { var podVolumeSpec *v1.VolumeSource @@ -279,7 +278,7 @@ func mountData(Ctx context.Context, config DockerConfig, pod v1.Pod, data interf } if podVolumeSpec != nil && podVolumeSpec.ConfigMap != nil { - podConfigMapDir := filepath.Join(wd+"/"+config.DataRootFolder+pod.Namespace+"-"+string(pod.UID)+"/", "configMaps/", vol.Name) + podConfigMapDir := filepath.Join(config.DataRootFolder+pod.Namespace+"-"+string(pod.UID), "configMaps", vol.Name) mode := os.FileMode(*podVolumeSpec.ConfigMap.DefaultMode) correctMountPath := "" @@ -330,7 +329,7 @@ func mountData(Ctx context.Context, config DockerConfig, pod v1.Pod, data interf } if podVolumeSpec != nil && podVolumeSpec.Secret != nil { mode := os.FileMode(*podVolumeSpec.Secret.DefaultMode) - podSecretDir := filepath.Join(wd+"/"+config.DataRootFolder+pod.Namespace+"-"+string(pod.UID)+"/", "secrets/", vol.Name) + podSecretDir := filepath.Join(config.DataRootFolder+pod.Namespace+"-"+string(pod.UID), "secrets", vol.Name) if mount.Data != nil { for key := range mount.Data { @@ -395,7 +394,7 @@ func mountData(Ctx context.Context, config DockerConfig, pod v1.Pod, data interf } } - edPath = filepath.Join(wd + "/" + config.DataRootFolder + pod.Namespace + "-" + string(pod.UID) + "/" + "emptyDirs/" + vol.Name) + edPath = filepath.Join(config.DataRootFolder+pod.Namespace+"-"+string(pod.UID), "emptyDirs", vol.Name) cmd := []string{"-p " + edPath} shell := exec2.ExecTask{ Command: "mkdir", @@ -424,5 +423,5 @@ func mountData(Ctx context.Context, config DockerConfig, pod v1.Pod, data interf } } - return nil, err + return nil, errors.New("Volume " + container.Name + " not found in pod spec") } diff --git a/pkg/docker/types.go b/pkg/docker/types.go index a6e8190..b1265b8 100644 --- a/pkg/docker/types.go +++ b/pkg/docker/types.go @@ -25,6 +25,11 @@ type DockerConfig struct { SingularityPrefix string `yaml:"SingularityPrefix"` SingularityPath string `yaml:"SingularityPath"` EnableProbes bool `yaml:"EnableProbes"` + FPGAEnabled bool `yaml:"FPGAEnabled"` + XilinxToolsPath string `yaml:"XilinxToolsPath"` + VitisPath string `yaml:"VitisPath"` + XRTPath string `yaml:"XRTPath"` + DockerNetworkSubnet []string `yaml:"DockerNetworkSubnet"` set bool } @@ -67,31 +72,32 @@ type RetrievedContainer struct { // InterLinkConfig holds the whole configuration type InterLinkConfig struct { - VKConfigPath string `yaml:"VKConfigPath"` - VKTokenFile string `yaml:"VKTokenFile"` - Interlinkurl string `yaml:"InterlinkURL"` - Sidecarurl string `yaml:"SidecarURL"` - Sbatchpath string `yaml:"SbatchPath"` - Scancelpath string `yaml:"ScancelPath"` - Squeuepath string `yaml:"SqueuePath"` - Interlinkport string `yaml:"InterlinkPort"` - Socket string `yaml:"Socket"` - Sidecarport string `yaml:"SidecarPort"` - Commandprefix string `yaml:"CommandPrefix"` - ExportPodData bool `yaml:"ExportPodData"` - DataRootFolder string `yaml:"DataRootFolder"` - ServiceAccount string `yaml:"ServiceAccount"` - Namespace string `yaml:"Namespace"` - Tsocks bool `yaml:"Tsocks"` - Tsockspath string `yaml:"TsocksPath"` - Tsocksconfig string `yaml:"TsocksConfig"` - Tsockslogin string `yaml:"TsocksLoginNode"` - BashPath string `yaml:"BashPath"` - VerboseLogging bool `yaml:"VerboseLogging"` - ErrorsOnlyLogging bool `yaml:"ErrorsOnlyLogging"` - PodIP string `yaml:"PodIP"` - SingularityPrefix string `yaml:"SingularityPrefix"` - set bool + VKConfigPath string `yaml:"VKConfigPath"` + VKTokenFile string `yaml:"VKTokenFile"` + Interlinkurl string `yaml:"InterlinkURL"` + Sidecarurl string `yaml:"SidecarURL"` + Sbatchpath string `yaml:"SbatchPath"` + Scancelpath string `yaml:"ScancelPath"` + Squeuepath string `yaml:"SqueuePath"` + Interlinkport string `yaml:"InterlinkPort"` + Socket string `yaml:"Socket"` + Sidecarport string `yaml:"SidecarPort"` + Commandprefix string `yaml:"CommandPrefix"` + ExportPodData bool `yaml:"ExportPodData"` + DataRootFolder string `yaml:"DataRootFolder"` + ServiceAccount string `yaml:"ServiceAccount"` + Namespace string `yaml:"Namespace"` + Tsocks bool `yaml:"Tsocks"` + Tsockspath string `yaml:"TsocksPath"` + Tsocksconfig string `yaml:"TsocksConfig"` + Tsockslogin string `yaml:"TsocksLoginNode"` + BashPath string `yaml:"BashPath"` + VerboseLogging bool `yaml:"VerboseLogging"` + ErrorsOnlyLogging bool `yaml:"ErrorsOnlyLogging"` + PodIP string `yaml:"PodIP"` + SingularityPrefix string `yaml:"SingularityPrefix"` + DockerNetworkSubnet []string `yaml:"DockerNetworkSubnet"` + set bool } // ContainerLogOpts is a struct in which it is possible to specify options to retrieve logs from the sidecar