From eb85e089453eb030ac769e86fbb7a64e080271a5 Mon Sep 17 00:00:00 2001 From: dhaval055 Date: Mon, 9 Oct 2023 18:59:56 +0000 Subject: [PATCH] orborus changes --- functions/k8s-confs/orborus-deployment.yaml | 7 - functions/onprem/orborus/orborus.go | 165 +++++++++++++++++++- 2 files changed, 161 insertions(+), 11 deletions(-) diff --git a/functions/k8s-confs/orborus-deployment.yaml b/functions/k8s-confs/orborus-deployment.yaml index cc157289..05ebfe08 100644 --- a/functions/k8s-confs/orborus-deployment.yaml +++ b/functions/k8s-confs/orborus-deployment.yaml @@ -24,10 +24,6 @@ spec: io.kompose.network/shuffle: "true" io.kompose.service: orborus spec: - # volumes: - # - name: docker-socket - # hostPath: - # path: /var/run/docker.sock containers: - env: - name: BASE_URL @@ -61,9 +57,6 @@ spec: imagePullPolicy: Never name: shuffle-orborus resources: {} - # volumeMounts: - # - name: docker-socket - # mountPath: /var/run/docker.sock hostname: shuffle-orborus restartPolicy: Always status: {} diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index adc66075..5a4d262d 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -36,7 +36,7 @@ import ( //"github.com/docker/docker/api/types/filters" dockerclient "github.com/docker/docker/client" - // uuid "github.com/satori/go.uuid" + uuid "github.com/satori/go.uuid" //"github.com/mackerelio/go-osstat/disk" "github.com/mackerelio/go-osstat/memory" @@ -613,6 +613,7 @@ func deployWorker(image string, identifier string, env []string, executionReques }, Spec: corev1.PodSpec{ RestartPolicy: "Never", + // once images is pushed, we can remove this NodeSelector: map[string]string{ "node": "master", }, @@ -629,13 +630,170 @@ func deployWorker(image string, identifier string, env []string, executionReques createdPod, err := clientset.CoreV1().Pods("shuffle").Create(context.Background(), pod, metav1.CreateOptions{}) if err != nil { fmt.Fprintf(os.Stderr, "Error creating pod: %v\n", err) - os.Exit(1) } fmt.Printf("Created pod %q in namespace %q\n", createdPod.Name, createdPod.Namespace) } else { - // docker part here + // Binds is the actual "-v" volume. + // Max 20% CPU every second + + //CPUQuota: 25000, + //CPUPeriod: 100000, + //CPUShares: 256, + hostConfig := &container.HostConfig{ + LogConfig: container.LogConfig{ + Type: "json-file", + Config: map[string]string{ + "max-size": "10m", + }, + }, + Resources: container.Resources{}, + } + + if len(os.Getenv("DOCKER_HOST")) == 0 { + if runtime.GOOS == "windows" { + hostConfig.Binds = []string{`\\.\pipe\docker_engine:\\.\pipe\docker_engine`} + } else { + hostConfig.Binds = []string{"/var/run/docker.sock:/var/run/docker.sock:rw"} + } + } + + hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:%s", containerId)) + + if strings.ToLower(cleanupEnv) == "true" { + hostConfig.AutoRemove = true + } + + config := &container.Config{ + Image: image, + Env: env, + } + + //var swarmConfig = os.Getenv("SHUFFLE_SWARM_CONFIG") + parsedUuid := uuid.NewV4() + if swarmConfig == "run" || swarmConfig == "swarm" { + // FIXME: Should we handle replies properly? + // In certain cases, a workflow may e.g. be aborted already. If it's aborted, that returns + // a 401 from the worker, which returns an error here + go sendWorkerRequest(executionRequest) + //sendWorkerRequest(executionRequest) + + //err := sendWorkerRequest(executionRequest) + //if err != nil { + // log.Printf("[ERROR] Failed worker request for %s: %s", executionRequest.ExecutionId, err) + + // if strings.Contains(fmt.Sprintf("%s", err), "connection refused") || strings.Contains(fmt.Sprintf("%s", err), "EOF") { + // workerImage := fmt.Sprintf("%s/%s/shuffle-worker:%s", baseimageregistry, baseimagename, workerVersion) + // deployServiceWorkers(workerImage) + + // time.Sleep(time.Duration(10) * time.Second) + // err = sendWorkerRequest(executionRequest) + // } + //} + + //if err == nil { + // // FIXME: Readd this? Removed for rerun reasons + // // executionIds = append(executionIds, executionRequest.ExecutionId) + //} + //}() + + return nil + } + + //log.Printf("[INFO] Identifier: %s", identifier) + cont, err := dockercli.ContainerCreate( + context.Background(), + config, + hostConfig, + nil, + nil, + identifier, + ) + + if err != nil { + if strings.Contains(fmt.Sprintf("%s", err), "Conflict. The container name ") { + identifier = fmt.Sprintf("%s-%s", identifier, parsedUuid) + //log.Printf("[INFO] 2 - Identifier: %s", identifier) + cont, err = dockercli.ContainerCreate( + context.Background(), + config, + hostConfig, + nil, + nil, + identifier, + ) + + if err != nil { + log.Printf("[ERROR] Container create error(2): %s", err) + return err + } + } else { + log.Printf("[ERROR] Container create error: %s", err) + return err + } + } + + containerStartOptions := types.ContainerStartOptions{} + err = dockercli.ContainerStart(context.Background(), cont.ID, containerStartOptions) + if err != nil { + // Trying to recreate and start WITHOUT network if it's possible. No extended checks. Old execution system (<0.9.30) + if strings.Contains(fmt.Sprintf("%s", err), "cannot join network") || strings.Contains(fmt.Sprintf("%s", err), "No such container") { + hostConfig.NetworkMode = "" + //container.NetworkMode(fmt.Sprintf("container:%s", containerId)) + cont, err = dockercli.ContainerCreate( + context.Background(), + config, + hostConfig, + nil, + nil, + identifier+"-2", + ) + if err != nil { + log.Printf("[ERROR] Failed to CREATE container (2): %s", err) + } + + err = dockercli.ContainerStart(context.Background(), cont.ID, containerStartOptions) + if err != nil { + log.Printf("[ERROR] Failed to start container (2): %s", err) + } + } else { + log.Printf("[ERROR] Failed initial container start. Quitting as this is NOT a simple network issue. Err: %s", err) + } + + if err != nil { + log.Printf("[ERROR] Failed to start worker container in environment %s: %s", environment, err) + return err + } else { + log.Printf("[INFO] Worker Container %s was created under environment %s for execution %s: docker logs %s", cont.ID, environment, executionRequest.ExecutionId, cont.ID) + } + + //stats, err := cli.ContainerInspect(context.Background(), containerName) + //if err != nil { + // log.Printf("Failed checking worker %s", containerName) + // return + //} + + //containerStatus := stats.ContainerJSONBase.State.Status + //if containerStatus != "running" { + // log.Printf("Status of %s is %s. Should be running. Will reset", containerName, containerStatus) + // err = stopWorker(containerName) + // if err != nil { + // log.Printf("Failed stopping worker %s", execution.ExecutionId) + // return + // } + + // err = deployWorke(cli, workerImage, containerName, env) + // if err != nil { + // log.Printf("Failed executing worker %s in state %s", execution.ExecutionId, containerStatus) + // return + // } + //} + } else { + log.Printf("[INFO] Worker Container %s was created under environment %s: docker logs %s", cont.ID, environment, cont.ID) + } + + return nil } return nil @@ -889,7 +1047,6 @@ func main() { log.Printf("[INFO] Running inside k8s cluster") } - ///////////////////////// startupDelay := os.Getenv("SHUFFLE_ORBORUS_STARTUP_DELAY") if len(startupDelay) > 0 {