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

more k8s stuff

Browse Source
This commit is contained in:
dhaval055
2023-09-29 18:14:06 +00:00
parent 288b8afab9
commit 7e0a73d007
19 changed files with 1199 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files
functions/onprem/orborus/Dockerfile
+7
View File
@@ -23,4 +23,11 @@ FROM alpine:3.15.0
RUN apk add --no-cache bash tzdata
COPY --from=builder /app/ /
ENV ENVIRONMENT_NAME=Shuffle
ENV BASE_URL=http://shuffle-backend:5001
ENV DOCKER_API_VERSION=1.39
ENV SHUFFLE_OPENSEARCH_URL=https://opensearch:9200
CMD ["./orborus"]
functions/onprem/orborus/orborus.go
+43 -185
View File
@@ -44,13 +44,13 @@ import (
//k8s deps
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/tools/clientcmd"
"k8s.io/client-go/util/homedir"
"k8s.io/client-go/rest"
"path/filepath"
"k8s.io/client-go/tools/clientcmd"
"k8s.io/client-go/util/homedir"
"path/filepath"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
// Starts jobs in bulk, so this could be increased
@@ -243,11 +243,11 @@ func deployServiceWorkers(image string) {
// this assumes that the machine should have at least 2 network
// interfaces. If not, we will use the default MTU.
// interface 1 is the loopback interface
// interface 2 is eth0, The eth0 interface inside a
// interface 2 is eth0, The eth0 interface inside a
// Docker container corresponds to the virtual Ethernet
// interface that connects the container to the docker0
log.Printf("[ERROR] Failed to get enough network interfaces")
} else {
} else {
// Get the preferred interface
for _, iface := range interfaces {
if strings.Contains(iface.Name, bridgeName) {
@@ -257,7 +257,7 @@ func deployServiceWorkers(image string) {
break
}
}
}
}
// Create the network options with the specified MTU
options := make(map[string]string)
@@ -579,205 +579,63 @@ func buildEnvVars(envMap map[string]string) []corev1.EnvVar {
}
func deployWorker(image string, identifier string, env []string, executionRequest shuffle.ExecutionRequest) error {
// Binds is the actual "-v" volume.
// Max 20% CPU every second
//CPUQuota: 25000,
//CPUPeriod: 100000,
//CPUShares: 256,
if os.Getenv("IS_KUBERNETES") == "true" {
log.Printf("IS_KUBERNETS", os.Getenv("IS_KUBERNETES"))
log.Printf("REGISTRY_URL", os.Getenv("REGISTRY_URL"))
fmt.Printf("ENV: %+v", env)
image = "shuffle-worker:v1" //hard coded image name to test locally
envMap := make(map[string]string)
for _, envStr := range env {
parts := strings.SplitN(envStr, "=", 2)
if len(parts) == 2 {
envMap[parts[0]] = parts[1]
if len(os.Getenv("REGISTRY_URL")) > 0 && os.Getenv("REGISTRY_URL") != "" {
env = append(env, fmt.Sprintf("REGISTRY_URL=%s", os.Getenv("REGISTRY_URL")))
env = append(env, fmt.Sprintf("IS_KUBERNETES=%s", os.Getenv("IS_KUBERNETES")))
}
}
clientset, err := getKubernetesClient()
if err != nil {
fmt.Println("[ERROR]Error getting kubernetes client:", err)
os.Exit(1)
}
image = "shuffle-worker:v1" //hard coded image name to test locally
envMap := make(map[string]string)
for _, envStr := range env {
parts := strings.SplitN(envStr, "=", 2)
if len(parts) == 2 {
envMap[parts[0]] = parts[1]
}
}
log.Printf("envMap", envMap)
clientset, err := getKubernetesClient()
if err != nil {
fmt.Println("[ERROR]Error getting kubernetes client:", err)
os.Exit(1)
}
pod := &corev1.Pod{
ObjectMeta: metav1.ObjectMeta{
Name: identifier,
Name: identifier,
Labels: map[string]string{"app": "shuffle-worker"},
},
Spec: corev1.PodSpec{
NodeSelector: map[string]string{
"node": "master",
},
Containers: []corev1.Container{
{
Name: identifier,
Image: image,
Env: buildEnvVars(envMap),
Env: buildEnvVars(envMap),
},
},
},
}
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 {
// 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)
// }
// docker part here
}
return nil
}
@@ -1026,7 +884,7 @@ func getKubernetesClient() (*kubernetes.Clientset, error) {
// Initial loop etc
func main() {
if isRunningInCluster(){
if isRunningInCluster() {
log.Printf("[INFO] Running inside k8s cluster")
}
@@ -1116,7 +974,7 @@ func main() {
ctx := context.Background()
// Run by default from now
//commenting for now as its stoppoing minikube
// zombiecheck(ctx, workerTimeout)
// zombiecheck(ctx, workerTimeout)
log.Printf("[INFO] Running towards %s (BASE_URL) with environment name %s", baseUrl, environment)