Merge branch '2.0.0' of https://github.com/Shuffle/Shuffle into flipBranch
This commit is contained in:
monilprajapati
@@ -1,15 +1,27 @@
|
||||
apiVersion: v1
|
||||
kind: Namespace
|
||||
metadata:
|
||||
name: shuffle
|
||||
|
||||
---
|
||||
|
||||
apiVersion: storage.k8s.io/v1
|
||||
kind: StorageClass
|
||||
metadata:
|
||||
name: shuffle-data
|
||||
namespace: shuffle
|
||||
provisioner: kubernetes.io/no-provisioner
|
||||
volumeBindingMode: WaitForFirstConsumer
|
||||
|
||||
---
|
||||
|
||||
apiVersion: v1
|
||||
metadata:
|
||||
namespace: shuffle
|
||||
creationTimestamp: null
|
||||
labels:
|
||||
io.kompose.service: backend-env
|
||||
name: env
|
||||
data:
|
||||
BACKEND_HOSTNAME: shuffle-backend
|
||||
BACKEND_PORT: "5001"
|
||||
@@ -51,11 +63,13 @@ data:
|
||||
SHUFFLE_OPENSEARCH_APIKEY: ""
|
||||
SHUFFLE_OPENSEARCH_CERTIFICATE_FILE: ""
|
||||
SHUFFLE_OPENSEARCH_CLOUDID: ""
|
||||
KUBERNETES_NAMESPACE: shuffle
|
||||
SHUFFLE_OPENSEARCH_INDEX_PREFIX: ""
|
||||
SHUFFLE_OPENSEARCH_PASSWORD: admin
|
||||
SHUFFLE_OPENSEARCH_PROXY: ""
|
||||
SHUFFLE_OPENSEARCH_SKIPSSL_VERIFY: "true"
|
||||
SHUFFLE_OPENSEARCH_URL: https://opensearch:9200
|
||||
SHUFFLE_MEMCACHED: shuffle-memcached:11211
|
||||
SHUFFLE_OPENSEARCH_USERNAME: admin
|
||||
SHUFFLE_ORBORUS_STARTUP_DELAY: "\t\t"
|
||||
SHUFFLE_PASS_APP_PROXY: "FALSE"
|
||||
@@ -68,11 +82,6 @@ data:
|
||||
REGISTRY_AUTH: "false"
|
||||
SHUFFLE_KUBERNETES_WORKER: "ghcr.io/shuffle/shuffle-worker:nightly"
|
||||
kind: ConfigMap
|
||||
metadata:
|
||||
creationTimestamp: null
|
||||
labels:
|
||||
io.kompose.service: backend-env
|
||||
name: env
|
||||
|
||||
---
|
||||
|
||||
@@ -107,25 +116,27 @@ roleRef:
|
||||
|
||||
---
|
||||
|
||||
apiVersion: v1
|
||||
kind: PersistentVolume
|
||||
metadata:
|
||||
name: shuffle-os-pv
|
||||
spec:
|
||||
capacity:
|
||||
storage: 10Gi # Adjust the storage size as per your requirements
|
||||
accessModes:
|
||||
- ReadWriteOnce # This allows read-write access to a single node
|
||||
persistentVolumeReclaimPolicy: Retain # Adjust the reclaim policy as per your needs
|
||||
storageClassName: shuffle-data # Set the desired storage class
|
||||
hostPath:
|
||||
path: /mnt/shuffle-data/open-search
|
||||
# apiVersion: v1
|
||||
# kind: PersistentVolume
|
||||
# metadata:
|
||||
# name: shuffle-os-pv
|
||||
# namespace: shuffle
|
||||
# spec:
|
||||
# capacity:
|
||||
# storage: 10Gi # Adjust the storage size as per your requirements
|
||||
# accessModes:
|
||||
# - ReadWriteOnce # This allows read-write access to a single node
|
||||
# persistentVolumeReclaimPolicy: Retain # Adjust the reclaim policy as per your needs
|
||||
# storageClassName: standard-rwo # Set the desired storage class
|
||||
# hostPath:
|
||||
# path: /mnt/shuffle-data/open-search
|
||||
|
||||
---
|
||||
# ---
|
||||
|
||||
apiVersion: v1
|
||||
kind: PersistentVolumeClaim
|
||||
metadata:
|
||||
namespace: shuffle
|
||||
creationTimestamp: null
|
||||
labels:
|
||||
io.kompose.service: opensearch-claim0
|
||||
@@ -133,7 +144,7 @@ metadata:
|
||||
spec:
|
||||
accessModes:
|
||||
- ReadWriteOnce
|
||||
storageClassName: shuffle-data
|
||||
storageClassName: standard-rwo
|
||||
resources:
|
||||
requests:
|
||||
storage: 500Mi
|
||||
@@ -143,6 +154,7 @@ status: {}
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
namespace: shuffle
|
||||
annotations:
|
||||
kompose.cmd: kompose convert -f docker-compose.yml
|
||||
kompose.version: 1.26.0 (40646f47)
|
||||
@@ -222,6 +234,7 @@ status: {}
|
||||
apiVersion: v1
|
||||
kind: Service
|
||||
metadata:
|
||||
namespace: shuffle
|
||||
annotations:
|
||||
kompose.cmd: kompose convert -f docker-compose.yml
|
||||
kompose.version: 1.26.0 (40646f47)
|
||||
@@ -241,51 +254,54 @@ status:
|
||||
|
||||
---
|
||||
|
||||
# apiVersion: v1
|
||||
# kind: PersistentVolume
|
||||
# metadata:
|
||||
# namespace: shuffle
|
||||
# name: shuffle-apps-pv
|
||||
# spec:
|
||||
# capacity:
|
||||
# storage: 5Gi
|
||||
# accessModes:
|
||||
# - ReadWriteOnce
|
||||
# persistentVolumeReclaimPolicy: Retain
|
||||
# storageClassName: shuffle-data
|
||||
# hostPath:
|
||||
# path: /mnt/shuffle-data/backend
|
||||
|
||||
# ---
|
||||
# apiVersion: v1
|
||||
# kind: PersistentVolume
|
||||
# metadata:
|
||||
# namespace: shuffle
|
||||
# name: shuffle-files-pv
|
||||
# spec:
|
||||
# capacity:
|
||||
# storage: 5Gi
|
||||
# accessModes:
|
||||
# - ReadWriteOnce
|
||||
# persistentVolumeReclaimPolicy: Retain
|
||||
# storageClassName: shuffle-data
|
||||
# hostPath:
|
||||
# path: /mnt/shuffle-data/backend
|
||||
|
||||
# ---
|
||||
|
||||
apiVersion: v1
|
||||
kind: PersistentVolume
|
||||
kind: PersistentVolumeClaim
|
||||
metadata:
|
||||
name: shuffle-apps-pv
|
||||
namespace: shuffle
|
||||
creationTimestamp: null
|
||||
labels:
|
||||
io.kompose.service: backend-files-claim
|
||||
name: backend-files-claim
|
||||
spec:
|
||||
capacity:
|
||||
storage: 5Gi
|
||||
accessModes:
|
||||
- ReadWriteOnce
|
||||
persistentVolumeReclaimPolicy: Retain
|
||||
storageClassName: shuffle-data
|
||||
hostPath:
|
||||
path: /mnt/shuffle-data/backend
|
||||
|
||||
---
|
||||
apiVersion: v1
|
||||
kind: PersistentVolume
|
||||
metadata:
|
||||
name: shuffle-files-pv
|
||||
spec:
|
||||
capacity:
|
||||
storage: 5Gi
|
||||
accessModes:
|
||||
- ReadWriteOnce
|
||||
persistentVolumeReclaimPolicy: Retain
|
||||
storageClassName: shuffle-data
|
||||
hostPath:
|
||||
path: /mnt/shuffle-data/backend
|
||||
|
||||
---
|
||||
|
||||
apiVersion: v1
|
||||
kind: PersistentVolumeClaim
|
||||
metadata:
|
||||
creationTimestamp: null
|
||||
labels:
|
||||
io.kompose.service: backend-files-claim
|
||||
name: backend-files-claim
|
||||
spec:
|
||||
accessModes:
|
||||
- ReadWriteOnce
|
||||
storageClassName: shuffle-data
|
||||
resources:
|
||||
requests:
|
||||
storage: 5Gi
|
||||
accessModes:
|
||||
- ReadWriteOnce
|
||||
storageClassName: standard-rwo
|
||||
resources:
|
||||
requests:
|
||||
storage: 5Gi
|
||||
# status: {}
|
||||
|
||||
---
|
||||
@@ -293,6 +309,7 @@ spec:
|
||||
apiVersion: v1
|
||||
kind: PersistentVolumeClaim
|
||||
metadata:
|
||||
namespace: shuffle
|
||||
creationTimestamp: null
|
||||
labels:
|
||||
io.kompose.service: backend-apps-claim
|
||||
@@ -300,7 +317,7 @@ metadata:
|
||||
spec:
|
||||
accessModes:
|
||||
- ReadWriteOnce
|
||||
storageClassName: shuffle-data
|
||||
storageClassName: standard-rwo
|
||||
resources:
|
||||
requests:
|
||||
storage: 5Gi
|
||||
@@ -311,6 +328,48 @@ spec:
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
name: shuffle-memcached
|
||||
namespace: shuffle
|
||||
spec:
|
||||
replicas: 1
|
||||
selector:
|
||||
matchLabels:
|
||||
app: shuffle-memcached
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app: shuffle-memcached
|
||||
spec:
|
||||
containers:
|
||||
- name: shuffle-memcached
|
||||
image: memcached:latest
|
||||
ports:
|
||||
- containerPort: 11211
|
||||
resources: {}
|
||||
restartPolicy: Always
|
||||
|
||||
|
||||
---
|
||||
|
||||
apiVersion: v1
|
||||
kind: Service
|
||||
metadata:
|
||||
namespace: shuffle
|
||||
name: shuffle-memcached
|
||||
spec:
|
||||
ports:
|
||||
- port: 11211
|
||||
targetPort: 11211
|
||||
selector:
|
||||
app: shuffle-memcached
|
||||
type: ClusterIP
|
||||
|
||||
---
|
||||
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
namespace: shuffle
|
||||
annotations:
|
||||
kompose.cmd: kompose convert -f docker-compose.yml
|
||||
kompose.version: 1.26.0 (40646f47)
|
||||
@@ -538,6 +597,11 @@ spec:
|
||||
configMapKeyRef:
|
||||
key: SHUFFLE_OPENSEARCH_APIKEY
|
||||
name: env
|
||||
- name: SHUFFLE_MEMCACHED
|
||||
valueFrom:
|
||||
configMapKeyRef:
|
||||
key: SHUFFLE_MEMCACHED
|
||||
name: env
|
||||
- name: SHUFFLE_OPENSEARCH_CERTIFICATE_FILE
|
||||
valueFrom:
|
||||
configMapKeyRef:
|
||||
@@ -642,6 +706,7 @@ status: {}
|
||||
apiVersion: v1
|
||||
kind: Service
|
||||
metadata:
|
||||
namespace: shuffle
|
||||
annotations:
|
||||
kompose.cmd: kompose convert -f docker-compose.yml
|
||||
kompose.version: 1.26.0 (40646f47)
|
||||
@@ -664,6 +729,7 @@ status:
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
namespace: shuffle
|
||||
annotations:
|
||||
kompose.cmd: kompose convert -f docker-compose.yml
|
||||
kompose.version: 1.26.0 (40646f47)
|
||||
@@ -705,6 +771,7 @@ status: {}
|
||||
apiVersion: v1
|
||||
kind: Service
|
||||
metadata:
|
||||
namespace: shuffle
|
||||
annotations:
|
||||
kompose.cmd: kompose convert -f docker-compose.yml
|
||||
kompose.version: 1.26.0 (40646f47)
|
||||
@@ -733,6 +800,7 @@ spec:
|
||||
apiVersion: apps/v1
|
||||
kind: Deployment
|
||||
metadata:
|
||||
namespace: shuffle
|
||||
annotations:
|
||||
kompose.cmd: kompose convert -f docker-compose.yml
|
||||
kompose.version: 1.26.0 (40646f47)
|
||||
@@ -779,6 +847,11 @@ spec:
|
||||
configMapKeyRef:
|
||||
key: IS_KUBERNETES
|
||||
name: env
|
||||
- name: KUBERNETES_NAMESPACE
|
||||
valueFrom:
|
||||
configMapKeyRef:
|
||||
key: KUBERNETES_NAMESPACE
|
||||
name: env
|
||||
- name: REGISTRY_URL
|
||||
valueFrom:
|
||||
configMapKeyRef:
|
||||
@@ -790,6 +863,11 @@ spec:
|
||||
key: SHUFFLE_KUBERNETES_WORKER
|
||||
name: env
|
||||
|
||||
- name: SHUFFLE_MEMCACHED
|
||||
valueFrom:
|
||||
configMapKeyRef:
|
||||
key: SHUFFLE_MEMCACHED
|
||||
name: env
|
||||
image: ghcr.io/shuffle/shuffle-orborus:nightly
|
||||
#imagePullPolicy: Never
|
||||
name: shuffle-orborus
|
||||
|
||||
+401
-297
@@ -49,11 +49,11 @@ import (
|
||||
//"github.com/mackerelio/go-osstat/memory"
|
||||
//"github.com/shirou/gopsutil/cpu"
|
||||
|
||||
appsv1 "k8s.io/api/apps/v1"
|
||||
corev1 "k8s.io/api/core/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
rbacv1 "k8s.io/api/rbac/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/util/intstr"
|
||||
|
||||
)
|
||||
|
||||
// Starts jobs in bulk, so this could be increased
|
||||
@@ -75,7 +75,6 @@ var isKubernetes = os.Getenv("IS_KUBERNETES")
|
||||
var kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE")
|
||||
var maxCPUPercent = 90
|
||||
|
||||
|
||||
// var baseimagename = "docker.pkg.github.com/shuffle/shuffle"
|
||||
// var baseimagename = "ghcr.io/frikky"
|
||||
// var baseimagename = "shuffle/shuffle"
|
||||
@@ -104,7 +103,7 @@ var memcached = os.Getenv("SHUFFLE_MEMCACHED")
|
||||
var tenzirUrl = os.Getenv("SHUFFLE_TENZIR_URL")
|
||||
|
||||
var executionIds = []string{}
|
||||
var namespacemade = false // For K8s
|
||||
var namespacemade = false // For K8s
|
||||
|
||||
var dockercli *dockerclient.Client
|
||||
var containerId string
|
||||
@@ -177,9 +176,26 @@ func getThisContainerId() {
|
||||
log.Printf(`[INFO] Started with containerId "%s"`, containerId)
|
||||
}
|
||||
|
||||
|
||||
func skipCheckInCleanup(name string) bool {
|
||||
return strings.HasPrefix(name, "backend") ||
|
||||
strings.HasPrefix(name, "shuffle-backend") ||
|
||||
strings.HasPrefix(name, "frontend") ||
|
||||
strings.HasPrefix(name, "shuffle-frontend") ||
|
||||
strings.HasPrefix(name, "orborus") ||
|
||||
strings.HasPrefix(name, "shuffle-orborus") ||
|
||||
strings.HasPrefix(name, "opensearch") ||
|
||||
strings.HasPrefix(name, "shuffle-opensearch") ||
|
||||
strings.HasPrefix(name, "memcached") ||
|
||||
strings.HasPrefix(name, "shuffle-memcached")
|
||||
}
|
||||
|
||||
func cleanupExistingNodes(ctx context.Context) error {
|
||||
|
||||
if isKubernetes == "true" {
|
||||
// of course, this doesn't clean up "nodes" but
|
||||
// rather pods, services, roles etc.
|
||||
|
||||
if kubernetesNamespace == "" {
|
||||
kubernetesNamespace = "default"
|
||||
}
|
||||
@@ -198,6 +214,12 @@ func cleanupExistingNodes(ctx context.Context) error {
|
||||
}
|
||||
|
||||
for _, pod := range pods.Items {
|
||||
// check if pod.Name starts with:
|
||||
// "backend-", "frontend-", "orborus-", "opensearch-" or "memcached-"
|
||||
if skipCheckInCleanup(pod.Name) {
|
||||
continue
|
||||
}
|
||||
|
||||
err := clientset.CoreV1().Pods(kubernetesNamespace).Delete(context.Background(), pod.Name, metav1.DeleteOptions{})
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed deleting pod %s: %s", pod.Name, err)
|
||||
@@ -212,17 +234,40 @@ func cleanupExistingNodes(ctx context.Context) error {
|
||||
}
|
||||
|
||||
for _, service := range services.Items {
|
||||
if skipCheckInCleanup(service.Name) {
|
||||
continue
|
||||
}
|
||||
|
||||
err := clientset.CoreV1().Services(kubernetesNamespace).Delete(context.Background(), service.Name, metav1.DeleteOptions{})
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed deleting service %s: %s", service.Name, err)
|
||||
}
|
||||
}
|
||||
|
||||
log.Printf("[INFO] Cleaned up all pods and services in namespace %s", kubernetesNamespace)
|
||||
deployments, err := clientset.AppsV1().Deployments(kubernetesNamespace).List(context.Background(), metav1.ListOptions{})
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed listing deployments: %s", err)
|
||||
return err
|
||||
}
|
||||
|
||||
for _, deployment := range deployments.Items {
|
||||
if skipCheckInCleanup(deployment.Name) {
|
||||
continue
|
||||
}
|
||||
|
||||
err := clientset.AppsV1().Deployments(kubernetesNamespace).Delete(context.Background(), deployment.Name, metav1.DeleteOptions{})
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed deleting deployment %s: %s", deployment.Name, err)
|
||||
}
|
||||
}
|
||||
|
||||
log.Printf("[INFO] Cleaned up all pods and services in namespace %s. Waiting 10 seconds for cleanup to reflect", kubernetesNamespace)
|
||||
|
||||
time.Sleep(10 * time.Second)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
serviceListOptions := types.ServiceListOptions{}
|
||||
services, err := dockercli.ServiceList(
|
||||
context.Background(),
|
||||
@@ -672,7 +717,6 @@ func buildEnvVars(envMap map[string]string) []corev1.EnvVar {
|
||||
return envVars
|
||||
}
|
||||
|
||||
|
||||
func handleBackendImageDownload(ctx context.Context, images string) error {
|
||||
// Should use docker to:
|
||||
// 1. Pull the image & tag it
|
||||
@@ -687,8 +731,7 @@ func handleBackendImageDownload(ctx context.Context, images string) error {
|
||||
log.Printf("[DEBUG] Should remove existing image (s): %s", images)
|
||||
|
||||
// Remove the image
|
||||
removeOptions := image.RemoveOptions{
|
||||
}
|
||||
removeOptions := image.RemoveOptions{}
|
||||
|
||||
for _, image := range strings.Split(images, ",") {
|
||||
image = strings.TrimSpace(image)
|
||||
@@ -708,6 +751,120 @@ func handleBackendImageDownload(ctx context.Context, images string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func fixk8sRoles() {
|
||||
clientset, _, err := shuffle.GetKubernetesClient()
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Error getting kubernetes client: %s", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
kubernetesNamespace := "default"
|
||||
|
||||
// Check if namespace exist as variable. If so, make it
|
||||
if len(os.Getenv("KUBERNETES_NAMESPACE")) > 0 {
|
||||
kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE")
|
||||
}
|
||||
|
||||
// fix roles
|
||||
// check if "service-creator" role is assigned to the service account "default"
|
||||
// roleBindingNames := []string{"service-creator-binding", "pod-creator-binding", "deployment-creator-binding"}
|
||||
serviceAccountName := "default"
|
||||
roleBindingName := "creator-all"
|
||||
|
||||
resourceTypes := []string{"services", "pods", "deployments"}
|
||||
|
||||
// Check if the RoleBinding exists
|
||||
roleBinding, err := clientset.RbacV1().RoleBindings(kubernetesNamespace).Get(context.TODO(), roleBindingName, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
log.Printf("[WARNING] Failed to get RoleBinding %s: %s", roleBindingName, err)
|
||||
|
||||
// create role and rolebinding
|
||||
role := &rbacv1.Role{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: roleBindingName,
|
||||
},
|
||||
Rules: []rbacv1.PolicyRule{
|
||||
{
|
||||
APIGroups: []string{"", "apps"},
|
||||
Resources: resourceTypes,
|
||||
Verbs: []string{"create", "list"},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
ctx := context.TODO()
|
||||
|
||||
_, err := clientset.RbacV1().Roles(kubernetesNamespace).Create(ctx, role, metav1.CreateOptions{})
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed to create Role %s: %s", roleBindingName, err)
|
||||
if !strings.Contains(fmt.Sprintf("%s", err), "already exists") {
|
||||
log.Printf("[INFO] role %s already exists", roleBindingName)
|
||||
}
|
||||
}
|
||||
|
||||
roleBinding := &rbacv1.RoleBinding{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: roleBindingName,
|
||||
},
|
||||
Subjects: []rbacv1.Subject{
|
||||
{
|
||||
Kind: "ServiceAccount",
|
||||
Name: serviceAccountName,
|
||||
Namespace: kubernetesNamespace,
|
||||
},
|
||||
},
|
||||
RoleRef: rbacv1.RoleRef{
|
||||
Kind: "Role",
|
||||
Name: roleBindingName,
|
||||
},
|
||||
}
|
||||
|
||||
_, err = clientset.RbacV1().RoleBindings(kubernetesNamespace).Create(ctx, roleBinding, metav1.CreateOptions{})
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed to create RoleBinding %s: %s", roleBindingName, err)
|
||||
if strings.Contains(fmt.Sprintf("%s", err), "already exists") {
|
||||
log.Printf("[INFO] rolebinding %s already exists", roleBindingName)
|
||||
}
|
||||
}
|
||||
|
||||
log.Printf("[INFO] Created Role %s and RoleBinding %s", roleBindingName, roleBindingName)
|
||||
} else {
|
||||
log.Printf("[INFO] RoleBinding %s exists", roleBindingName)
|
||||
}
|
||||
|
||||
// Check if the RoleBinding is assigned to the service account
|
||||
var found bool
|
||||
for _, subject := range roleBinding.Subjects {
|
||||
if subject.Kind == "ServiceAccount" && subject.Name == serviceAccountName {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if !found {
|
||||
log.Printf("[WARNING] Service account %s is not assigned to RoleBinding %s\n", serviceAccountName, roleBindingName)
|
||||
// assign the service account to the rolebinding
|
||||
roleBinding.Subjects = append(roleBinding.Subjects, rbacv1.Subject{
|
||||
Kind: "ServiceAccount",
|
||||
Name: serviceAccountName,
|
||||
Namespace: kubernetesNamespace,
|
||||
})
|
||||
|
||||
ctx := context.TODO()
|
||||
|
||||
_, err := clientset.RbacV1().RoleBindings(kubernetesNamespace).Update(ctx, roleBinding, metav1.UpdateOptions{})
|
||||
if err != nil {
|
||||
log.Printf("[ERROR](ns - %s) Failed to update RoleBinding %s: %s", kubernetesNamespace, roleBindingName, err)
|
||||
if !strings.Contains(fmt.Sprintf("%s", err), "already exists") {
|
||||
log.Printf("[INFO] rolebinding %s already exists", roleBindingName)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
func int32Ptr(i int32) *int32 { return &i }
|
||||
|
||||
func deployK8sWorker(image string, identifier string, env []string) error {
|
||||
env = append(env, fmt.Sprintf("IS_KUBERNETES=true"))
|
||||
env = append(env, fmt.Sprintf("KUBERNETES_NAMESPACE=%s", os.Getenv("KUBERNETES_NAMESPACE")))
|
||||
@@ -744,7 +901,6 @@ func deployK8sWorker(image string, identifier string, env []string) error {
|
||||
// use k8s downward API to find it if we are in a pod
|
||||
}
|
||||
|
||||
|
||||
// Check if namespace exist as variable. If so, make it
|
||||
if len(os.Getenv("KUBERNETES_NAMESPACE")) > 0 && !namespacemade {
|
||||
kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE")
|
||||
@@ -770,6 +926,7 @@ func deployK8sWorker(image string, identifier string, env []string) error {
|
||||
|
||||
env = append(env, fmt.Sprintf("BASE_URL=%s", baseUrl))
|
||||
env = append(env, fmt.Sprintf("SHUFFLE_SWARM_CONFIG=%s", swarmConfig))
|
||||
env = append(env, fmt.Sprintf("WORKER_HOSTNAME=%s", "shuffle-workers"))
|
||||
|
||||
if len(kubernetesNamespace) == 0 {
|
||||
foundNamespace, err := shuffle.GetKubernetesNamespace()
|
||||
@@ -829,66 +986,131 @@ func deployK8sWorker(image string, identifier string, env []string) error {
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// While testing:
|
||||
// kubectl delete pods --all --all-namespaces; kubectl delete services --all --all-namespaces
|
||||
pod := &corev1.Pod{
|
||||
// pod := &corev1.Pod{
|
||||
// ObjectMeta: metav1.ObjectMeta{
|
||||
// Name: identifier,
|
||||
// Labels: containerLabels,
|
||||
// },
|
||||
// Spec: corev1.PodSpec{
|
||||
// RestartPolicy: "Never",
|
||||
// // DNSPolicy: "Default",
|
||||
// DNSPolicy: corev1.DNSClusterFirst,
|
||||
// // NodeSelector: map[string]string{
|
||||
// // "node": "master",
|
||||
// // },
|
||||
// Containers: []corev1.Container{
|
||||
// containerAttachment,
|
||||
// },
|
||||
// },
|
||||
// }
|
||||
|
||||
// // Check if running on ARM or x86 to download the correct image
|
||||
|
||||
// // Get current pod's network so we can make the pod in it
|
||||
|
||||
// _, err = clientset.CoreV1().Pods(kubernetesNamespace).List(context.Background(), metav1.ListOptions{})
|
||||
// if err != nil {
|
||||
// log.Printf("[ERROR] Failed listing pods: %s", err)
|
||||
// }
|
||||
|
||||
// createdPod, err := clientset.CoreV1().Pods(kubernetesNamespace).Create(context.Background(), pod, metav1.CreateOptions{})
|
||||
// if err != nil {
|
||||
// //log.Printf("[ERROR] Failed creating pod: %v", err)
|
||||
// return err
|
||||
// }
|
||||
|
||||
// log.Printf("[INFO] Created pod %q in namespace %q\n", createdPod.Name, createdPod.Namespace)
|
||||
|
||||
// // kubectl expose pod shuffle-workers --type=LoadBalancer --port=33333
|
||||
// service := &corev1.Service{
|
||||
// ObjectMeta: metav1.ObjectMeta{
|
||||
// Name: identifier,
|
||||
// },
|
||||
// Spec: corev1.ServiceSpec{
|
||||
// Selector: map[string]string{
|
||||
// "container": "shuffle-workers",
|
||||
// },
|
||||
// Ports: []corev1.ServicePort{
|
||||
// {
|
||||
// Protocol: "TCP",
|
||||
// Port: 33333,
|
||||
// TargetPort: intstr.FromInt(33333),
|
||||
// },
|
||||
// },
|
||||
// Type: corev1.ServiceTypeLoadBalancer,
|
||||
// },
|
||||
// }
|
||||
|
||||
// _, err = clientset.CoreV1().Services(kubernetesNamespace).Create(context.TODO(), service, metav1.CreateOptions{})
|
||||
// if err != nil {
|
||||
// log.Printf("[ERROR] Failed creating service: %v", err)
|
||||
// return err
|
||||
// }
|
||||
|
||||
replicaNumberStr := os.Getenv("SHUFFLE_SCALE_REPLICAS")
|
||||
replicaNumber := 1
|
||||
if len(replicaNumberStr) > 0 {
|
||||
tmpInt, err := strconv.Atoi(replicaNumberStr)
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] %s is not a valid number for replication", replicaNumberStr)
|
||||
} else {
|
||||
replicaNumber = tmpInt
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
replicaNumberInt32 := int32(replicaNumber)
|
||||
|
||||
deployment := &appsv1.Deployment{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: identifier,
|
||||
Labels: containerLabels,
|
||||
Name: identifier,
|
||||
},
|
||||
Spec: corev1.PodSpec{
|
||||
RestartPolicy: "Never",
|
||||
// DNSPolicy: "Default",
|
||||
DNSPolicy: corev1.DNSClusterFirst,
|
||||
// NodeSelector: map[string]string{
|
||||
// "node": "master",
|
||||
// },
|
||||
Containers: []corev1.Container{
|
||||
containerAttachment,
|
||||
Spec: appsv1.DeploymentSpec{
|
||||
Replicas: int32Ptr(replicaNumberInt32),
|
||||
Selector: &metav1.LabelSelector{
|
||||
MatchLabels: containerLabels,
|
||||
},
|
||||
Template: corev1.PodTemplateSpec{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Labels: containerLabels,
|
||||
},
|
||||
Spec: corev1.PodSpec{
|
||||
Containers: []corev1.Container{
|
||||
containerAttachment,
|
||||
},
|
||||
DNSPolicy: corev1.DNSClusterFirst,
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
// Check if running on ARM or x86 to download the correct image
|
||||
|
||||
// Get current pod's network so we can make the pod in it
|
||||
|
||||
_, err = clientset.CoreV1().Pods(kubernetesNamespace).List(context.Background(), metav1.ListOptions{})
|
||||
_, err = clientset.AppsV1().Deployments(kubernetesNamespace).Create(context.Background(), deployment, metav1.CreateOptions{})
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed listing pods: %s", err)
|
||||
}
|
||||
|
||||
|
||||
createdPod, err := clientset.CoreV1().Pods(kubernetesNamespace).Create(context.Background(), pod, metav1.CreateOptions{})
|
||||
if err != nil {
|
||||
//log.Printf("[ERROR] Failed creating pod: %v", err)
|
||||
log.Printf("[ERROR] Failed creating deployment: %v", err)
|
||||
return err
|
||||
}
|
||||
|
||||
log.Printf("[INFO] Created pod %q in namespace %q\n", createdPod.Name, createdPod.Namespace)
|
||||
|
||||
// kubectl expose pod shuffle-workers --type=LoadBalancer --port=33333
|
||||
// kubectl expose deployment shuffle-workers --type=NodePort --port=33333 --target-port=33333
|
||||
service := &corev1.Service{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: identifier,
|
||||
},
|
||||
Spec: corev1.ServiceSpec{
|
||||
Selector: map[string]string{
|
||||
"container": "shuffle-workers",
|
||||
},
|
||||
Selector: containerLabels,
|
||||
Ports: []corev1.ServicePort{
|
||||
{
|
||||
Protocol: "TCP",
|
||||
Port: 33333,
|
||||
Protocol: "TCP",
|
||||
Port: 33333,
|
||||
TargetPort: intstr.FromInt(33333),
|
||||
},
|
||||
},
|
||||
Type: corev1.ServiceTypeLoadBalancer,
|
||||
Type: corev1.ServiceTypeNodePort,
|
||||
},
|
||||
}
|
||||
|
||||
_, err = clientset.CoreV1().Services(kubernetesNamespace).Create(context.TODO(), service, metav1.CreateOptions{})
|
||||
_, err = clientset.CoreV1().Services(kubernetesNamespace).Create(context.Background(), service, metav1.CreateOptions{})
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed creating service: %v", err)
|
||||
return err
|
||||
@@ -897,7 +1119,6 @@ func deployK8sWorker(image string, identifier string, env []string) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
func deployWorker(image string, identifier string, env []string, executionRequest shuffle.ExecutionRequest) error {
|
||||
if len(os.Getenv("REGISTRY_URL")) > 0 && os.Getenv("REGISTRY_URL") != "" {
|
||||
env = append(env, fmt.Sprintf("REGISTRY_URL=%s", os.Getenv("REGISTRY_URL")))
|
||||
@@ -1299,8 +1520,7 @@ func getOrborusStats(ctx context.Context) shuffle.OrborusStats {
|
||||
newStats.MaxMemory = int(pers.MemTotal)
|
||||
}
|
||||
|
||||
|
||||
// Get list of all running containers
|
||||
// Get list of all running containers
|
||||
containers, err := dockercli.ContainerList(ctx, container.ListOptions{})
|
||||
|
||||
if err != nil {
|
||||
@@ -1414,10 +1634,6 @@ func getOrborusStats(ctx context.Context) shuffle.OrborusStats {
|
||||
return newStats
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
func sendRemoveRequest(client *http.Client, toBeRemoved shuffle.ExecutionRequestWrapper, baseUrl, environment, auth, org string, sleepTime int) error {
|
||||
confirmUrl := fmt.Sprintf("%s/api/v1/workflows/queue/confirm", baseUrl)
|
||||
|
||||
@@ -1496,112 +1712,7 @@ func main() {
|
||||
}
|
||||
|
||||
if isKubernetes == "true" {
|
||||
clientset, _, err := shuffle.GetKubernetesClient()
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Error getting kubernetes client: %s", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
kubernetesNamespace := "default"
|
||||
|
||||
// Check if namespace exist as variable. If so, make it
|
||||
if len(os.Getenv("KUBERNETES_NAMESPACE")) > 0 && !namespacemade {
|
||||
kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE")
|
||||
}
|
||||
|
||||
// fix roles
|
||||
// check if "service-creator" role is assigned to the service account "default"
|
||||
roleBindingName := "service-creator-binding"
|
||||
serviceAccountName := "default"
|
||||
// Check if the RoleBinding exists
|
||||
roleBinding, err := clientset.RbacV1().RoleBindings(kubernetesNamespace).Get(context.TODO(), roleBindingName, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
log.Printf("[WARNING] Failed to get RoleBinding %s: %s", roleBindingName, err)
|
||||
// create role and rolebinding
|
||||
role := &rbacv1.Role{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: roleBindingName,
|
||||
},
|
||||
Rules: []rbacv1.PolicyRule{
|
||||
{
|
||||
APIGroups: []string{""},
|
||||
Resources: []string{"services"},
|
||||
Verbs: []string{"create"},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
ctx := context.TODO()
|
||||
|
||||
_, err := clientset.RbacV1().Roles(kubernetesNamespace).Create(ctx, role, metav1.CreateOptions{})
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed to create Role %s: %s", roleBindingName, err)
|
||||
if !strings.Contains(fmt.Sprintf("%s", err), "already exists") {
|
||||
log.Printf("[INFO] role %s already exists", roleBindingName)
|
||||
}
|
||||
}
|
||||
|
||||
roleBinding := &rbacv1.RoleBinding{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Name: roleBindingName,
|
||||
},
|
||||
Subjects: []rbacv1.Subject{
|
||||
{
|
||||
Kind: "ServiceAccount",
|
||||
Name: serviceAccountName,
|
||||
Namespace: kubernetesNamespace,
|
||||
},
|
||||
},
|
||||
RoleRef: rbacv1.RoleRef{
|
||||
Kind: "Role",
|
||||
Name: roleBindingName,
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
|
||||
_, err = clientset.RbacV1().RoleBindings(kubernetesNamespace).Create(ctx, roleBinding, metav1.CreateOptions{})
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed to create RoleBinding %s: %s", roleBindingName, err)
|
||||
if !strings.Contains(fmt.Sprintf("%s", err), "already exists") {
|
||||
log.Printf("[INFO] rolebinding %s already exists", roleBindingName)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
log.Printf("[INFO] Created Role %s and RoleBinding %s", roleBindingName, roleBindingName)
|
||||
} else {
|
||||
log.Printf("[INFO] RoleBinding %s exists", roleBindingName)
|
||||
}
|
||||
|
||||
// Check if the RoleBinding is assigned to the service account
|
||||
var found bool
|
||||
for _, subject := range roleBinding.Subjects {
|
||||
if subject.Kind == "ServiceAccount" && subject.Name == serviceAccountName {
|
||||
found = true
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if !found {
|
||||
log.Printf("[WARNING] Service account %s is not assigned to RoleBinding %s\n", serviceAccountName, roleBindingName)
|
||||
// assign the service account to the rolebinding
|
||||
roleBinding.Subjects = append(roleBinding.Subjects, rbacv1.Subject{
|
||||
Kind: "ServiceAccount",
|
||||
Name: serviceAccountName,
|
||||
Namespace: kubernetesNamespace,
|
||||
})
|
||||
|
||||
ctx := context.TODO()
|
||||
|
||||
_, err := clientset.RbacV1().RoleBindings(kubernetesNamespace).Update(ctx, roleBinding, metav1.UpdateOptions{})
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed to update RoleBinding %s: %s", roleBindingName, err)
|
||||
if !strings.Contains(fmt.Sprintf("%s", err), "already exists") {
|
||||
log.Printf("[INFO] rolebinding %s already exists", roleBindingName)
|
||||
}
|
||||
}
|
||||
}
|
||||
fixk8sRoles()
|
||||
}
|
||||
|
||||
startupDelay := os.Getenv("SHUFFLE_ORBORUS_STARTUP_DELAY")
|
||||
@@ -1879,7 +1990,6 @@ func main() {
|
||||
continue
|
||||
}
|
||||
|
||||
|
||||
if hasStarted && len(executionRequests.Data) > 0 {
|
||||
//log.Printf("[INFO] Body: %s", string(body))
|
||||
// Type string `json:"type"`
|
||||
@@ -1957,7 +2067,7 @@ func main() {
|
||||
log.Printf("[WARNING] Throttle - Cutting down requests from %d to %d (MAX: %d, CUR: %d)", len(executionRequests.Data), allowed, maxConcurrency, executionCount)
|
||||
executionRequests.Data = executionRequests.Data[0:allowed]
|
||||
}
|
||||
} else if (swarmControlMode && (swarmConfig == "run" || swarmConfig == "swarm")) {
|
||||
} else if swarmControlMode && (swarmConfig == "run" || swarmConfig == "swarm") {
|
||||
if len(executionRequests.Data) > 50 {
|
||||
executionRequests.Data = executionRequests.Data[0:50]
|
||||
}
|
||||
@@ -2106,7 +2216,6 @@ func main() {
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// func deployPipeline(image, identifier, command string) error {
|
||||
// if isKubernetes == "true" {
|
||||
// return errors.New("Kubernetes not implemented")
|
||||
@@ -2131,7 +2240,6 @@ func main() {
|
||||
// envVariables := []string{
|
||||
// }
|
||||
|
||||
|
||||
// // Add volume binds for storage
|
||||
// // Want read/write with full access for the container
|
||||
// //sourceFolder := "/Users/frikky/git/shuffle/shuffle-database"
|
||||
@@ -2170,7 +2278,6 @@ func main() {
|
||||
// "shuffle": "shuffle",
|
||||
// }
|
||||
|
||||
|
||||
// cont, err := dockercli.ContainerCreate(
|
||||
// ctx,
|
||||
// config,
|
||||
@@ -2256,8 +2363,6 @@ func main() {
|
||||
// return nil
|
||||
// }
|
||||
|
||||
|
||||
|
||||
// Tenzir command samples
|
||||
// docker pull ghcr.io/dominiklohmann/tenzir-arm64:latest
|
||||
// docker tag ghcr.io/dominiklohmann/tenzir-arm64:latest tenzir/tenzir:latest
|
||||
@@ -2268,11 +2373,11 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
|
||||
|
||||
if tenzirUrl == "" {
|
||||
tenzirUrl = "http://localhost:5160"
|
||||
log.Printf("[WARNING] SHUFFLE_TENZIR_URL not set, falling back to default URL: %s",tenzirUrl)
|
||||
log.Printf("[WARNING] SHUFFLE_TENZIR_URL not set, falling back to default URL: %s", tenzirUrl)
|
||||
}
|
||||
|
||||
err := deployTenzirNode()
|
||||
if err != nil{
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] failed to deploy the pipeline, reason: %s", err)
|
||||
return err
|
||||
}
|
||||
@@ -2322,7 +2427,7 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
|
||||
log.Printf("[INFO] successfully stopped the Pipeline: %s", pipelineId)
|
||||
}
|
||||
|
||||
} else if incRequest.Type == "PIPELINE_START" {
|
||||
} else if incRequest.Type == "PIPELINE_START" {
|
||||
log.Printf("[INFO] Should start the pipeline %#v", identifier)
|
||||
pipelineId, err := searchPipeline(identifier)
|
||||
if err != nil {
|
||||
@@ -2351,157 +2456,157 @@ func handlePipeline(incRequest shuffle.ExecutionRequest) error {
|
||||
}
|
||||
|
||||
func deployTenzirNode() error {
|
||||
if isKubernetes == "true" {
|
||||
return errors.New("kubernetes not implemented")
|
||||
}
|
||||
if isKubernetes == "true" {
|
||||
return errors.New("kubernetes not implemented")
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
cacheKey := "tenzir-key"
|
||||
ctx := context.Background()
|
||||
cacheKey := "tenzir-key"
|
||||
|
||||
imageName := "tenzir/tenzir:latest"
|
||||
containerName := "tenzir-node"
|
||||
containerStartOptions := container.StartOptions{}
|
||||
imageName := "tenzir/tenzir:latest"
|
||||
containerName := "tenzir-node"
|
||||
containerStartOptions := container.StartOptions{}
|
||||
|
||||
_, err := shuffle.GetCache(ctx, cacheKey)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
_, err := shuffle.GetCache(ctx, cacheKey)
|
||||
if err == nil {
|
||||
return nil
|
||||
}
|
||||
|
||||
containerInfo, err := dockercli.ContainerInspect(ctx, containerName)
|
||||
if err != nil {
|
||||
if dockerclient.IsErrNotFound(err) {
|
||||
containerInfo, err := dockercli.ContainerInspect(ctx, containerName)
|
||||
if err != nil {
|
||||
if dockerclient.IsErrNotFound(err) {
|
||||
|
||||
// Check if image exists
|
||||
_, _, err := dockercli.ImageInspectWithRaw(ctx, imageName)
|
||||
if dockerclient.IsErrNotFound(err) {
|
||||
log.Printf("[DEBUG] pulling image %s", imageName)
|
||||
pullOptions := image.PullOptions{}
|
||||
out, err := dockercli.ImagePull(ctx, imageName, pullOptions)
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed to pull the Tenzir image: %s", err)
|
||||
return err
|
||||
}
|
||||
defer out.Close()
|
||||
// Check if image exists
|
||||
_, _, err := dockercli.ImageInspectWithRaw(ctx, imageName)
|
||||
if dockerclient.IsErrNotFound(err) {
|
||||
log.Printf("[DEBUG] pulling image %s", imageName)
|
||||
pullOptions := image.PullOptions{}
|
||||
out, err := dockercli.ImagePull(ctx, imageName, pullOptions)
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed to pull the Tenzir image: %s", err)
|
||||
return err
|
||||
}
|
||||
defer out.Close()
|
||||
|
||||
io.Copy(io.Discard, out)
|
||||
} else if err != nil {
|
||||
return err
|
||||
}
|
||||
io.Copy(io.Discard, out)
|
||||
} else if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = createAndStartTenzirNode(ctx, containerName, imageName, containerStartOptions)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
if !containerInfo.State.Running {
|
||||
log.Printf("[DEBUG] Tenzir Node exists but is not running")
|
||||
err := dockercli.ContainerStart(ctx, containerName, containerStartOptions)
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed to start Tenzir Node container: %v", err)
|
||||
return err
|
||||
}
|
||||
err = createAndStartTenzirNode(ctx, containerName, imageName, containerStartOptions)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
return err
|
||||
}
|
||||
} else {
|
||||
if !containerInfo.State.Running {
|
||||
log.Printf("[DEBUG] Tenzir Node exists but is not running")
|
||||
err := dockercli.ContainerStart(ctx, containerName, containerStartOptions)
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed to start Tenzir Node container: %v", err)
|
||||
return err
|
||||
}
|
||||
|
||||
log.Printf("[INFO] Waiting for Tenzir to become available ...")
|
||||
err = checkTenzirNode()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
log.Printf("[INFO] Waiting for Tenzir to become available ...")
|
||||
err = checkTenzirNode()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
tenzirStatus := struct {
|
||||
ContainerStatus string `json:"container_status"`
|
||||
}{
|
||||
ContainerStatus: "running",
|
||||
}
|
||||
tenzirStatus := struct {
|
||||
ContainerStatus string `json:"container_status"`
|
||||
}{
|
||||
ContainerStatus: "running",
|
||||
}
|
||||
|
||||
cacheData, err := json.Marshal(tenzirStatus)
|
||||
if err != nil {
|
||||
log.Printf("[WARNING] Failed marshalling execution: %s", err)
|
||||
}
|
||||
err = shuffle.SetCache(ctx, cacheKey, cacheData, 1)
|
||||
if err != nil {
|
||||
log.Printf("[WARNING] Failed updating cache for tenzir: %s", err)
|
||||
}
|
||||
cacheData, err := json.Marshal(tenzirStatus)
|
||||
if err != nil {
|
||||
log.Printf("[WARNING] Failed marshalling execution: %s", err)
|
||||
}
|
||||
err = shuffle.SetCache(ctx, cacheKey, cacheData, 1)
|
||||
if err != nil {
|
||||
log.Printf("[WARNING] Failed updating cache for tenzir: %s", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
return nil
|
||||
}
|
||||
|
||||
func checkTenzirNode() error {
|
||||
retries := 20
|
||||
retryInterval := 3 * time.Second
|
||||
url := fmt.Sprintf("%s/api/v0/ping",tenzirUrl)
|
||||
retries := 20
|
||||
retryInterval := 3 * time.Second
|
||||
url := fmt.Sprintf("%s/api/v0/ping", tenzirUrl)
|
||||
forwardMethod := "POST"
|
||||
|
||||
client := http.Client{}
|
||||
req, err := http.NewRequest(forwardMethod, url, nil)
|
||||
client := http.Client{}
|
||||
req, err := http.NewRequest(forwardMethod, url, nil)
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed to create HTTP request: %s", err)
|
||||
return err
|
||||
}
|
||||
|
||||
for i := 0; i < retries; i++ {
|
||||
resp, err := client.Do(req)
|
||||
if err == nil && resp.StatusCode == http.StatusOK {
|
||||
return nil
|
||||
}
|
||||
time.Sleep(retryInterval)
|
||||
}
|
||||
for i := 0; i < retries; i++ {
|
||||
resp, err := client.Do(req)
|
||||
if err == nil && resp.StatusCode == http.StatusOK {
|
||||
return nil
|
||||
}
|
||||
time.Sleep(retryInterval)
|
||||
}
|
||||
|
||||
return fmt.Errorf("tenzir node is not available")
|
||||
return fmt.Errorf("tenzir node is not available")
|
||||
}
|
||||
|
||||
func createAndStartTenzirNode(ctx context.Context, containerName, imageName string, containerStartOptions container.StartOptions) error {
|
||||
healthconfig := &container.HealthConfig{
|
||||
Test: []string{"tenzir --connection-timeout=30s --connection-retry-delay=1s 'api /ping'"},
|
||||
Interval: 30 * time.Second,
|
||||
Retries: 1,
|
||||
}
|
||||
healthconfig := &container.HealthConfig{
|
||||
Test: []string{"tenzir --connection-timeout=30s --connection-retry-delay=1s 'api /ping'"},
|
||||
Interval: 30 * time.Second,
|
||||
Retries: 1,
|
||||
}
|
||||
|
||||
config := &container.Config{
|
||||
Cmd: []string{"--commands=web server --mode=dev --bind=0.0.0.0"},
|
||||
Image: imageName,
|
||||
Healthcheck: healthconfig,
|
||||
ExposedPorts: nat.PortSet{"5160/tcp": struct{}{}},
|
||||
Entrypoint: []string{containerName},
|
||||
}
|
||||
config := &container.Config{
|
||||
Cmd: []string{"--commands=web server --mode=dev --bind=0.0.0.0"},
|
||||
Image: imageName,
|
||||
Healthcheck: healthconfig,
|
||||
ExposedPorts: nat.PortSet{"5160/tcp": struct{}{}},
|
||||
Entrypoint: []string{containerName},
|
||||
}
|
||||
|
||||
hostConfig := &container.HostConfig{
|
||||
PortBindings: nat.PortMap{
|
||||
"5160/tcp": []nat.PortBinding{{HostPort: "5160"}},
|
||||
},
|
||||
Mounts: []mount.Mount{
|
||||
{
|
||||
Type: mount.TypeVolume,
|
||||
Source: containerName,
|
||||
Target: "/var/lib/tenzir/",
|
||||
},
|
||||
},
|
||||
VolumeDriver: "local",
|
||||
}
|
||||
_, err := dockercli.ContainerCreate(ctx, config, hostConfig, nil, nil, containerName)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
hostConfig := &container.HostConfig{
|
||||
PortBindings: nat.PortMap{
|
||||
"5160/tcp": []nat.PortBinding{{HostPort: "5160"}},
|
||||
},
|
||||
Mounts: []mount.Mount{
|
||||
{
|
||||
Type: mount.TypeVolume,
|
||||
Source: containerName,
|
||||
Target: "/var/lib/tenzir/",
|
||||
},
|
||||
},
|
||||
VolumeDriver: "local",
|
||||
}
|
||||
_, err := dockercli.ContainerCreate(ctx, config, hostConfig, nil, nil, containerName)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
err = dockercli.ContainerStart(ctx, containerName, containerStartOptions)
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed to start Tenzir Node container: %v", err)
|
||||
return err
|
||||
}
|
||||
log.Printf("[INFO] Tenzir Node container started successfully")
|
||||
err = dockercli.ContainerStart(ctx, containerName, containerStartOptions)
|
||||
if err != nil {
|
||||
log.Printf("[ERROR] Failed to start Tenzir Node container: %v", err)
|
||||
return err
|
||||
}
|
||||
log.Printf("[INFO] Tenzir Node container started successfully")
|
||||
|
||||
log.Printf("[INFO] Waiting for Tenzir to become available ...")
|
||||
err = checkTenzirNode()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
log.Printf("[INFO] Successfully deployed Tenzir Node !")
|
||||
log.Printf("[INFO] Waiting for Tenzir to become available ...")
|
||||
err = checkTenzirNode()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
log.Printf("[INFO] Successfully deployed Tenzir Node !")
|
||||
|
||||
return nil
|
||||
return nil
|
||||
}
|
||||
|
||||
func createPipeline(command, identifier string) (string, error) {
|
||||
@@ -2509,7 +2614,7 @@ func createPipeline(command, identifier string) (string, error) {
|
||||
toBeDeleted := false
|
||||
pipelineId, err := searchPipeline(identifier)
|
||||
|
||||
url := fmt.Sprintf("%s/api/v0/pipeline/create", tenzirUrl)
|
||||
url := fmt.Sprintf("%s/api/v0/pipeline/create", tenzirUrl)
|
||||
forwardMethod := "POST"
|
||||
|
||||
if err != nil {
|
||||
@@ -2522,7 +2627,7 @@ func createPipeline(command, identifier string) (string, error) {
|
||||
log.Printf("[INFO] an existing pipeline found with ID: %s. it will be deleted", pipelineId)
|
||||
toBeDeleted = true
|
||||
}
|
||||
if strings.Contains(command, "shuffler.io") {
|
||||
if strings.Contains(command, "shuffler.io") {
|
||||
|
||||
} else {
|
||||
var scheme string
|
||||
@@ -2615,7 +2720,7 @@ func createPipeline(command, identifier string) (string, error) {
|
||||
|
||||
func updatePipelineState(pipelineId, action string) (string, error) {
|
||||
|
||||
url := fmt.Sprintf("%s/api/v0/pipeline/update", tenzirUrl)
|
||||
url := fmt.Sprintf("%s/api/v0/pipeline/update", tenzirUrl)
|
||||
forwardMethod := "POST"
|
||||
|
||||
requestBody := map[string]interface{}{
|
||||
@@ -2684,7 +2789,7 @@ func deletePipeline(pipelineId string) error {
|
||||
"id": pipelineId,
|
||||
}
|
||||
|
||||
url := fmt.Sprintf("%s/api/v0/pipeline/delete", tenzirUrl)
|
||||
url := fmt.Sprintf("%s/api/v0/pipeline/delete", tenzirUrl)
|
||||
forwardMethod := "POST"
|
||||
|
||||
requestBodyJSON, err := json.Marshal(requestBody)
|
||||
@@ -2732,7 +2837,7 @@ func searchPipeline(identifier string) (string, error) {
|
||||
|
||||
var reqBody []byte
|
||||
|
||||
url := fmt.Sprintf("%s/api/v0/pipeline/list", tenzirUrl)
|
||||
url := fmt.Sprintf("%s/api/v0/pipeline/list", tenzirUrl)
|
||||
|
||||
resp, err := http.Post(url, "application/json", bytes.NewBuffer(reqBody))
|
||||
if err != nil {
|
||||
@@ -2817,10 +2922,9 @@ func searchPipeline(identifier string) (string, error) {
|
||||
func getRunningWorkers(ctx context.Context, workerTimeout int) int {
|
||||
//log.Printf("[DEBUG] Getting running workers with API version %s", dockerApiVersion)
|
||||
counter := 0
|
||||
if isKubernetes == "true" {
|
||||
if isKubernetes == "true" {
|
||||
log.Printf("[INFO] Getting running workers in kubernetes")
|
||||
|
||||
|
||||
thresholdTime := time.Now().Add(time.Duration(-workerTimeout) * time.Second)
|
||||
|
||||
clientset, _, err := shuffle.GetKubernetesClient()
|
||||
@@ -3128,7 +3232,7 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest, image string,
|
||||
|
||||
debugCommand := fmt.Sprintf("docker service logs shuffle-workers 2>&1 -f | grep %s", workflowExecution.ExecutionId)
|
||||
if isKubernetes == "true" {
|
||||
debugCommand = fmt.Sprintf("kubectl logs -n %s %s | grep %s", kubernetesNamespace, identifier, workflowExecution.ExecutionId)
|
||||
debugCommand = fmt.Sprintf("kubectl logs -n %s container=shuffle-worker | grep %s", kubernetesNamespace, workflowExecution.ExecutionId)
|
||||
}
|
||||
|
||||
log.Printf("[DEBUG] Ran worker from request with execution ID: %s. Worker URL: %s. DEBUGGING:\n%s", workflowExecution.ExecutionId, streamUrl, debugCommand)
|
||||
|
||||
Reference in New Issue
Block a user