marker/shuffle-cracked
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

orborus changes

Browse Source
This commit is contained in:
dhaval055
2023-10-09 18:59:56 +00:00
parent bc089034a4
commit eb85e08945
2 changed files with 161 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
functions/k8s-confs/orborus-deployment.yaml
-7
View File
@@ -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: {}
functions/onprem/orborus/orborus.go
+161 -4
View File
@@ -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 {