From 5e7e120b8982c4b2b2d803682f5599dec6db1a6a Mon Sep 17 00:00:00 2001 From: Aditya <60684641+0x0elliot@users.noreply.github.com> Date: Thu, 13 Jun 2024 02:48:39 +0530 Subject: [PATCH 1/7] fix: working my way towards a stable k8s release --- functions/onprem/orborus/orborus.go | 307 ++++++++++++++-------------- 1 file changed, 158 insertions(+), 149 deletions(-) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index eddcfd71..0660dfd2 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -659,6 +659,161 @@ func handleBackendImageDownload(ctx context.Context, images string) error { return nil } +func deployK8sWorker(image string, identifier string, env []string, executionRequest shuffle.ExecutionRequest) error { + env = append(env, fmt.Sprintf("IS_KUBERNETES=true")) + env = append(env, fmt.Sprintf("KUBERNETES_NAMESPACE=%s", os.Getenv("KUBERNETES_NAMESPACE"))) + + if len(os.Getenv("KUBERNETES_SERVICE_HOST")) > 0 { + env = append(env, fmt.Sprintf("KUBERNETES_SERVICE_HOST=%s", os.Getenv("KUBERNETES_SERVICE_HOST"))) + } + + if len(os.Getenv("KUBERNETES_SERVICE_PORT")) > 0 { + env = append(env, fmt.Sprintf("KUBERNETES_SERVICE_PORT=%s", os.Getenv("KUBERNETES_SERVICE_PORT"))) + } + + clientset, _, err := shuffle.GetKubernetesClient() + if err != nil { + log.Printf("[ERROR] Error getting kubernetes client:", err) + return err + } + + + //env = append(env, fmt.Sprintf("KUBERNETES_CONFIG=%s", config.String())) + + // FIXME: When a service account is used, the account is also mounted in the pod + // The volume mount location is: + // /var/run/secrets/kubernetes.io/serviceaccount + + // Look for if there is a default service account in use + if len(os.Getenv("KUBERNETES_SERVICE_ACCOUNT")) > 0 { + log.Printf("[DEBUG] Using Kubernetes service account %s", os.Getenv("KUBERNETES_SERVICE_ACCOUNT")) + env = append(env, fmt.Sprintf("KUBERNETES_SERVICE_ACCOUNT=%s", os.Getenv("KUBERNETES_SERVICE_ACCOUNT"))) + + // 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") + + // Make the namespace + namespace := &corev1.Namespace{ + ObjectMeta: metav1.ObjectMeta{ + Name: os.Getenv("KUBERNETES_NAMESPACE"), + }, + } + + _, err := clientset.CoreV1().Namespaces().Create(context.Background(), namespace, metav1.CreateOptions{}) + if err != nil { + if !strings.Contains(strings.ToLower(fmt.Sprintf("%s", err)), "already exists") { + log.Printf("[ERROR] Failed creating Kubernetes namespace: %s", err) + } else { + namespacemade = true + } + } else { + namespacemade = true + } + } + + if len(kubernetesNamespace) == 0 { + foundNamespace, err := shuffle.GetKubernetesNamespace() + if err != nil { + //log.Printf("[ERROR] Failed getting Kubernetes namespace: %s", err) + } + + if len(foundNamespace) > 0 { + kubernetesNamespace = foundNamespace + os.Setenv("KUBERNETES_NAMESPACE", kubernetesNamespace) + } + } + + if len(kubernetesNamespace) == 0 { + kubernetesNamespace = "default" + } + + kubernetesImage := os.Getenv("SHUFFLE_KUBERNETES_WORKER") + if len(kubernetesImage) == 0 { + kubernetesImage = image + } + log.Printf("[DEBUG] Using Kubernetes worker image '%s'", kubernetesImage) + // 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] + } + } + + containerLabels := map[string]string{ + "container": "shuffle-worker", + } + + containerAttachment := corev1.Container{ + Name: identifier, + Image: kubernetesImage, + Env: buildEnvVars(envMap), + + //ImagePullPolicy: "Never", + ImagePullPolicy: corev1.PullIfNotPresent, + } + + podname := shuffle.GetPodName() + + ctx := context.Background() + + if len(podname) > 0 { + _, err := shuffle.GetCurrentPodNetworkConfig(ctx, clientset, kubernetesNamespace, podname) + if err != nil { + log.Printf("[ERROR] Failed getting current pod network: %s", err) + } else { + log.Printf("[DEBUG] Current pod found!") + // currentPodStatus = k8s.io/api/core/v1.PodStatus + } + } + + + // While testing: + // kubectl delete pods --all --all-namespaces; kubectl delete services --all --all-namespaces + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Name: identifier, + Labels: containerLabels, + }, + Spec: corev1.PodSpec{ + RestartPolicy: "Never", + DNSPolicy: "Default", + // 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) + 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") != "" { @@ -666,158 +821,12 @@ func deployWorker(image string, identifier string, env []string, executionReques } if isKubernetes == "true" { - env = append(env, fmt.Sprintf("IS_KUBERNETES=true")) - env = append(env, fmt.Sprintf("KUBERNETES_NAMESPACE=%s", os.Getenv("KUBERNETES_NAMESPACE"))) - - if len(os.Getenv("KUBERNETES_SERVICE_HOST")) > 0 { - env = append(env, fmt.Sprintf("KUBERNETES_SERVICE_HOST=%s", os.Getenv("KUBERNETES_SERVICE_HOST"))) - } - - if len(os.Getenv("KUBERNETES_SERVICE_PORT")) > 0 { - env = append(env, fmt.Sprintf("KUBERNETES_SERVICE_PORT=%s", os.Getenv("KUBERNETES_SERVICE_PORT"))) - } - - clientset, _, err := shuffle.GetKubernetesClient() + err := deployK8sWorker(image, identifier, env, executionRequest) if err != nil { - log.Printf("[ERROR] Error getting kubernetes client:", err) - return err + log.Printf("[ERROR] Failed deploying Kubernetes worker: %s", err) } - - //env = append(env, fmt.Sprintf("KUBERNETES_CONFIG=%s", config.String())) - - // FIXME: When a service account is used, the account is also mounted in the pod - // The volume mount location is: - // /var/run/secrets/kubernetes.io/serviceaccount - - // Look for if there is a default service account in use - if len(os.Getenv("KUBERNETES_SERVICE_ACCOUNT")) > 0 { - log.Printf("[DEBUG] Using Kubernetes service account %s", os.Getenv("KUBERNETES_SERVICE_ACCOUNT")) - env = append(env, fmt.Sprintf("KUBERNETES_SERVICE_ACCOUNT=%s", os.Getenv("KUBERNETES_SERVICE_ACCOUNT"))) - - // 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") - - // Make the namespace - namespace := &corev1.Namespace{ - ObjectMeta: metav1.ObjectMeta{ - Name: os.Getenv("KUBERNETES_NAMESPACE"), - }, - } - - _, err := clientset.CoreV1().Namespaces().Create(context.Background(), namespace, metav1.CreateOptions{}) - if err != nil { - if !strings.Contains(strings.ToLower(fmt.Sprintf("%s", err)), "already exists") { - log.Printf("[ERROR] Failed creating Kubernetes namespace: %s", err) - } else { - namespacemade = true - } - } else { - namespacemade = true - } - } - - if len(kubernetesNamespace) == 0 { - foundNamespace, err := shuffle.GetKubernetesNamespace() - if err != nil { - //log.Printf("[ERROR] Failed getting Kubernetes namespace: %s", err) - } - - if len(foundNamespace) > 0 { - kubernetesNamespace = foundNamespace - os.Setenv("KUBERNETES_NAMESPACE", kubernetesNamespace) - } - } - - if len(kubernetesNamespace) == 0 { - kubernetesNamespace = "default" - } - - kubernetesImage := os.Getenv("SHUFFLE_KUBERNETES_WORKER") - if len(kubernetesImage) == 0 { - kubernetesImage = image - } - log.Printf("[DEBUG] Using Kubernetes worker image '%s'", kubernetesImage) - // 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] - } - } - - containerLabels := map[string]string{ - "container": "shuffle-worker", - } - - containerAttachment := corev1.Container{ - Name: identifier, - Image: kubernetesImage, - Env: buildEnvVars(envMap), - - //ImagePullPolicy: "Never", - ImagePullPolicy: corev1.PullIfNotPresent, - } - - podname := shuffle.GetPodName() - - ctx := context.Background() - - if len(podname) > 0 { - currentPodStatus, err := shuffle.GetCurrentPodNetworkConfig(ctx, clientset, kubernetesNamespace, podname) - if err != nil { - log.Printf("[ERROR] Failed getting current pod network: %s", err) - } else { - log.Printf("[DEBUG] Current pod found!") - // currentPodStatus = k8s.io/api/core/v1.PodStatus - } - } - - - // While testing: - // kubectl delete pods --all --all-namespaces; kubectl delete services --all --all-namespaces - pod := &corev1.Pod{ - ObjectMeta: metav1.ObjectMeta{ - Name: identifier, - Labels: containerLabels, - }, - Spec: corev1.PodSpec{ - RestartPolicy: "Never", - DNSPolicy: "Default", - // 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 - - networks, 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) - return nil + return err } // Binds is the actual "-v" volume. From 824756a6efb69354ea8b3c0520f5477e45d1d068 Mon Sep 17 00:00:00 2001 From: Aditya <60684641+0x0elliot@users.noreply.github.com> Date: Fri, 14 Jun 2024 05:52:58 +0530 Subject: [PATCH 2/7] fix: k8s support enhancements --- functions/onprem/orborus/orborus.go | 126 +++++++++++++++++++++------- 1 file changed, 98 insertions(+), 28 deletions(-) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 0660dfd2..a7e1031b 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -50,6 +50,7 @@ import ( corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/util/intstr" ) // Starts jobs in bulk, so this could be increased @@ -659,7 +660,7 @@ func handleBackendImageDownload(ctx context.Context, images string) error { return nil } -func deployK8sWorker(image string, identifier string, env []string, executionRequest shuffle.ExecutionRequest) error { +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"))) @@ -677,7 +678,6 @@ func deployK8sWorker(image string, identifier string, env []string, executionReq return err } - //env = append(env, fmt.Sprintf("KUBERNETES_CONFIG=%s", config.String())) // FIXME: When a service account is used, the account is also mounted in the pod @@ -716,6 +716,9 @@ func deployK8sWorker(image string, identifier string, env []string, executionReq } } + env = append(env, fmt.Sprintf("BASE_URL=%s", baseUrl)) + env = append(env, fmt.Sprintf("SHUFFLE_SWARM_CONFIG=%s", swarmConfig)) + if len(kubernetesNamespace) == 0 { foundNamespace, err := shuffle.GetKubernetesNamespace() if err != nil { @@ -811,6 +814,33 @@ func deployK8sWorker(image string, identifier string, env []string, executionReq } 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-worker", + }, + 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 + } + return nil } @@ -820,14 +850,14 @@ func deployWorker(image string, identifier string, env []string, executionReques env = append(env, fmt.Sprintf("REGISTRY_URL=%s", os.Getenv("REGISTRY_URL"))) } - if isKubernetes == "true" { - err := deployK8sWorker(image, identifier, env, executionRequest) - if err != nil { - log.Printf("[ERROR] Failed deploying Kubernetes worker: %s", err) - } + // if isKubernetes == "true" { + // err := deployK8sWorker(image, identifier, env, executionRequest) + // if err != nil { + // log.Printf("[ERROR] Failed deploying Kubernetes worker: %s", err) + // } - return err - } + // return err + // } // Binds is the actual "-v" volume. // Max 20% CPU every second @@ -853,23 +883,25 @@ func deployWorker(image string, identifier string, env []string, executionReques } } - hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:%s", containerId)) - if strings.ToLower(cleanupEnv) != "false" { - hostConfig.AutoRemove = true - } - config := &container.Config{ Image: image, Env: env, } + if isKubernetes != "true" { + hostConfig.NetworkMode = container.NetworkMode(fmt.Sprintf("container:%s", containerId)) + if strings.ToLower(cleanupEnv) != "false" { + hostConfig.AutoRemove = true + } + } + //var swarmConfig = os.Getenv("SHUFFLE_SWARM_CONFIG") parsedUuid := uuid.NewV4() - if swarmConfig == "run" || swarmConfig == "swarm" { + if swarmConfig == "run" || swarmConfig == "swarm" || isKubernetes == "true" { // 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) + go sendWorkerRequest(executionRequest, image, env) return nil } @@ -1510,16 +1542,26 @@ func main() { workerImage = newWorkerImage } - if swarmConfig == "run" || swarmConfig == "swarm" { - checkSwarmService(ctx) + if swarmConfig == "run" || swarmConfig == "swarm" || isKubernetes == "true" { - log.Printf("[DEBUG] Cleaning up containers from previous run") - cleanupExistingNodes(ctx) - time.Sleep(time.Duration(5) * time.Second) + if isKubernetes != "true" { + checkSwarmService(ctx) + log.Printf("[DEBUG] Cleaning up containers from previous run") + cleanupExistingNodes(ctx) + time.Sleep(time.Duration(5) * time.Second) + } log.Printf("[DEBUG] Deploying worker image %s to swarm", workerImage) - deployServiceWorkers(workerImage) - log.Printf("[DEBUG] Waiting 45 seconds to ensure workers are deployed. Run: \"docker service ls\" for more info") + + runString := "Run: \"docker service ls\" for more info" + + if isKubernetes != "true" { + deployServiceWorkers(workerImage) + } else { + deployK8sWorker(workerImage, "shuffle-workers", []string{}) + runString = "Run: \"kubectl get pods\" for more info" + } + log.Printf("[DEBUG] Waiting 45 seconds to ensure workers are deployed. %s", runString) time.Sleep(time.Duration(45) * time.Second) //deployServiceWorkers(workerImage) @@ -1529,7 +1571,12 @@ func main() { client := shuffle.GetExternalClient(baseUrl) fullUrl := fmt.Sprintf("%s/api/v1/workflows/queue", baseUrl) - log.Printf("[INFO] Finished configuring docker environment. Connecting to %s", fullUrl) + + if isKubernetes == "true" { + log.Printf("[INFO] Finished configuring kubernetes environment. Connecting to %s", fullUrl) + } else { + log.Printf("[INFO] Finished configuring docker environment. Connecting to %s", fullUrl) + } forwardData := bytes.NewBuffer([]byte{}) forwardMethod := "POST" @@ -2794,7 +2841,7 @@ func zombiecheck(ctx context.Context, workerTimeout int) error { return nil } -func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error { +func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest, image string, env []string) error { parsedRequest := shuffle.OrborusExecutionRequest{ ExecutionId: workflowExecution.ExecutionId, Authorization: workflowExecution.Authorization, @@ -2827,6 +2874,15 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error { streamUrl = fmt.Sprintf("%s:33333/api/v1/execute", parsedBaseurl) } + identifier := "shuffle-workers" + + if isKubernetes == "true" { + if shuffle.IsRunningInCluster() { + log.Printf("[INFO] Running in Kubernetes cluster") + // try getting the k8s worker server url + } + } + if len(workerServerUrl) > 0 { streamUrl = fmt.Sprintf("%s:33333/api/v1/execute", workerServerUrl) } @@ -2851,7 +2907,12 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error { if len(newWorkerImage) > 0 { workerImage = newWorkerImage } - deployServiceWorkers(workerImage) + + if isKubernetes == "true" { + deployK8sWorker(workerImage, identifier, env) + } else { + deployServiceWorkers(workerImage) + } time.Sleep(time.Duration(10) * time.Second) //err = sendWorkerRequest(executionRequest) @@ -2870,7 +2931,11 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error { workerImage = newWorkerImage } - deployServiceWorkers(workerImage) + if isKubernetes == "true" { + deployK8sWorker(workerImage, identifier, env) + } else { + deployServiceWorkers(workerImage) + } time.Sleep(time.Duration(10) * time.Second) //err = sendWorkerRequest(executionRequest) @@ -2899,6 +2964,11 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error { _ = body - log.Printf("[DEBUG] Ran worker from request with execution ID: %s. Worker URL: %s. DEBUGGING:\ndocker service logs shuffle-workers 2>&1 -f | grep %s", workflowExecution.ExecutionId, streamUrl, workflowExecution.ExecutionId) + 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) + } + + log.Printf("[DEBUG] Ran worker from request with execution ID: %s. Worker URL: %s. DEBUGGING:\n%s", workflowExecution.ExecutionId, streamUrl, debugCommand) return nil } From 5322599c0ae5b56864e2eb2aac2102b9328077f3 Mon Sep 17 00:00:00 2001 From: Aditya <60684641+0x0elliot@users.noreply.github.com> Date: Fri, 21 Jun 2024 06:21:40 +0530 Subject: [PATCH 3/7] feat[k8s-keep-alive: making it work --- functions/onprem/orborus/orborus.go | 121 +++++++++++++++++++++++++++- 1 file changed, 118 insertions(+), 3 deletions(-) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index a7e1031b..e23a715b 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -50,7 +50,9 @@ import ( corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + rbacv1 "k8s.io/api/rbac/v1" "k8s.io/apimachinery/pkg/util/intstr" + ) // Starts jobs in bulk, so this could be increased @@ -666,7 +668,11 @@ func deployK8sWorker(image string, identifier string, env []string) error { if len(os.Getenv("KUBERNETES_SERVICE_HOST")) > 0 { env = append(env, fmt.Sprintf("KUBERNETES_SERVICE_HOST=%s", os.Getenv("KUBERNETES_SERVICE_HOST"))) - } + } + + if len(os.Getenv("SHUFFLE_MEMCACHED")) > 0 { + env = append(env, fmt.Sprintf("SHUFFLE_MEMCACHED=%s", os.Getenv("SHUFFLE_MEMCACHED"))) + } if len(os.Getenv("KUBERNETES_SERVICE_PORT")) > 0 { env = append(env, fmt.Sprintf("KUBERNETES_SERVICE_PORT=%s", os.Getenv("KUBERNETES_SERVICE_PORT"))) @@ -787,7 +793,8 @@ func deployK8sWorker(image string, identifier string, env []string) error { }, Spec: corev1.PodSpec{ RestartPolicy: "Never", - DNSPolicy: "Default", + // DNSPolicy: "Default", + DNSPolicy: corev1.DNSClusterFirst, // NodeSelector: map[string]string{ // "node": "master", // }, @@ -1442,6 +1449,115 @@ func main() { log.Printf("[INFO] Running inside k8s cluster") } + if isKubernetes == "true" { + clientset, _, err := shuffle.GetKubernetesClient() + if err != nil { + log.Printf("[ERROR] Error getting kubernetes client:", 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) + } + } + } + } + startupDelay := os.Getenv("SHUFFLE_ORBORUS_STARTUP_DELAY") if len(startupDelay) > 0 { log.Printf("[DEBUG] Setting startup delay to %#v", startupDelay) @@ -1543,7 +1659,6 @@ func main() { } if swarmConfig == "run" || swarmConfig == "swarm" || isKubernetes == "true" { - if isKubernetes != "true" { checkSwarmService(ctx) log.Printf("[DEBUG] Cleaning up containers from previous run") From ad9743af10927e63866504c72bd307d4fae98a8c Mon Sep 17 00:00:00 2001 From: Frikky Date: Fri, 21 Jun 2024 15:12:40 +0200 Subject: [PATCH 4/7] Fixed workflow editing things --- frontend/src/components/EditWorkflow.jsx | 75 ++++++++++------- frontend/src/components/Oauth2Auth.jsx | 18 ++-- frontend/src/views/Admin.jsx | 2 +- frontend/src/views/AngularWorkflow.jsx | 102 +++++++++++++++++++++-- frontend/src/views/Workflows.jsx | 13 +-- 5 files changed, 153 insertions(+), 57 deletions(-) diff --git a/frontend/src/components/EditWorkflow.jsx b/frontend/src/components/EditWorkflow.jsx index 4b4aa3d5..cc4a33b4 100644 --- a/frontend/src/components/EditWorkflow.jsx +++ b/frontend/src/components/EditWorkflow.jsx @@ -338,26 +338,6 @@ const EditWorkflow = (props) => { autoFocus fullWidth /> -
- { - setDescription(event.target.value) - }} - InputProps={{ - style: { - color: "white", - }, - }} - maxRows={4} - color="primary" - defaultValue={innerWorkflow.description} - placeholder="Description" - multiline - label="Description" - margin="dense" - fullWidth - /> -
{usecases !== null && usecases !== undefined && usecases.length > 0 ? @@ -425,27 +405,56 @@ const EditWorkflow = (props) => { fullWidth value={newWorkflowTags} onChange={(chip) => { - console.log("Chip: ", chip) - //newWorkflowTags.push(chip); setNewWorkflowTags(chip); }} + onBlur={(event) => { + if (event.target.value.length === 0) { + return + } + + if (newWorkflowTags.includes(event.target.value)) { + return + } + + newWorkflowTags.push(event.target.value) + setNewWorkflowTags(newWorkflowTags) + + setUpdate(Math.random()) + }} onAdd={(chip) => { - newWorkflowTags.push(chip); - setNewWorkflowTags(newWorkflowTags); + newWorkflowTags.push(chip) + setNewWorkflowTags(newWorkflowTags) }} onDelete={(chip, index) => { console.log("Deleting: ", chip, index) - newWorkflowTags.splice(index, 1); - setNewWorkflowTags(newWorkflowTags); - setUpdate(Math.random()); + newWorkflowTags.splice(index, 1) + setNewWorkflowTags(newWorkflowTags) + setUpdate(Math.random()) }} />
{showMoreClicked === true ? - - - +
+ { + setDescription(event.target.value) + }} + InputProps={{ + style: { + color: "white", + }, + }} + multiLine + rows={3} + color="primary" + defaultValue={innerWorkflow.description} + placeholder="Description" + multiline + label="Description" + margin="dense" + fullWidth + />
@@ -564,6 +573,8 @@ const EditWorkflow = (props) => { fullWidth /> + + MSSP Suborg Distribution (beta - contact support@shuffler.io for more info) @@ -669,6 +680,7 @@ const EditWorkflow = (props) => { } + Input fields @@ -781,6 +793,7 @@ const EditWorkflow = (props) => { + Git Backup Repository @@ -931,7 +944,7 @@ const EditWorkflow = (props) => { - +
: null} diff --git a/frontend/src/components/Oauth2Auth.jsx b/frontend/src/components/Oauth2Auth.jsx index f0c20a64..e9e794b2 100755 --- a/frontend/src/components/Oauth2Auth.jsx +++ b/frontend/src/components/Oauth2Auth.jsx @@ -471,32 +471,26 @@ const AuthenticationOauth2 = (props) => { state += `%26refresh_uri%3d${authentication_url}`; } - // No prompt forcing - //var url = `${authenticationType.redirect_uri}?client_id=${client_id}&redirect_uri=${redirectUri}&response_type=code&prompt=login&scope=${resources}&state=${state}&access_type=offline`; + // FIXME: Should this be =consent? var defaultPrompt = "login" if (prompt !== undefined && prompt !== null && prompt.length > 0) { - defaultPrompt = prompt + defaultPrompt = prompt } var url = `${authenticationType.redirect_uri}?client_id=${client_id}&redirect_uri=${redirectUri}&response_type=code&prompt=${defaultPrompt}&scope=${resources}&state=${state}&access_type=offline`; - if (admin_consent === true) { console.log("Running Oauth2 WITH admin consent") //url = `${authenticationType.redirect_uri}?client_id=${client_id}&redirect_uri=${redirectUri}&response_type=code&prompt=consent&scope=${resources}&state=${state}&access_type=offline`; url = `${authenticationType.redirect_uri}?client_id=${client_id}&redirect_uri=${redirectUri}&response_type=code&prompt=admin_consent&scope=${resources}&state=${state}&access_type=offline`; } - // Force new consent + console.log("OAUTH2 URL: ", url) + + // Force new consent //const url = `${authenticationType.redirect_uri}?client_id=${client_id}&redirect_uri=${redirectUri}&response_type=code&scope=${resources}&prompt=consent&state=${state}&access_type=offline`; - // Admin consent + // Admin consent //const url = `https://accounts.zoho.com/oauth/v2/auth?response_type=code&client_id=${client_id}&scope=AaaServer.profile.Read&redirect_uri=${redirectUri}&prompt=consent` - - // &resource=https%3A%2F%2Fgraph.microsoft.com& - - // FIXME: Awful, but works for prototyping - // How can we get a callback properly realtime? - // How can we properly try-catch without breaks on error? try { var newwin = window.open(url, "", "width=582,height=700"); //console.log(newwin) diff --git a/frontend/src/views/Admin.jsx b/frontend/src/views/Admin.jsx index 8546e887..f9f85761 100755 --- a/frontend/src/views/Admin.jsx +++ b/frontend/src/views/Admin.jsx @@ -3745,7 +3745,7 @@ If you're interested, please let me know a time that works for you, or set up a aria-label="disabled tabs example" > Edit Details /> - Org Limits & Cloud Sync /> + Limits & Cloud Sync /> Priorities /> Billing & Stats /> Branding (Beta) /> diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index 0a17d93d..5df415e8 100755 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -5294,7 +5294,59 @@ const releaseToConnectLabel = "Release to Connect" } setTimeout(() => { - setSelectedTriggerIndex(trigger_index); + if (trigger_index !== -1) { + const trigger = workflow.triggers[trigger_index] + if (trigger !== undefined && trigger !== null) { + + // Autofixer + if (trigger.trigger_type === "USERINPUT") { + const relevantparams = [ + "alertinfo", + "options", + "type", + "email", + "sms", + "subflow", + ] + var foundparams = 0 + for (var paramkey in trigger.parameters) { + if (relevantparams.includes(trigger.parameters[paramkey].name)) { + foundparams++ + } + } + + if (foundparams < 6) { + trigger.parameters = [{ + name: "alertinfo", + value: "Do you want to continue the workflow? Start parameters: $exec", + },{ + name: "options", + value: "boolean", + }, + { + name: "type", + value: "subflow", + }, + { + name: "email", + value: "test@test.com", + }, + { + name: "sms", + value: "0000000", + }, + { + name: "subflow", + value: "", + }] + + workflow.triggers[trigger_index].parameters = trigger.parameters + } + } + } + } + + setSelectedTriggerIndex(trigger_index) setSelectedTrigger(data) setSelectedActionEnvironment(data.env) }, 25) @@ -12382,7 +12434,7 @@ const releaseToConnectLabel = "Release to Connect" setSubworkflow(e.target.value); // Sets the startnode - if (e.target.value.id !== workflow.id) { + if (e.target.value.id !== workflow.id && e.target.value.id.length > 0 ) { console.log("WORKFLOW: ", e.target.value); const startnode = e.target.value.actions.find((action) => action.id === e.target.value.start); @@ -14808,6 +14860,10 @@ const releaseToConnectLabel = "Release to Connect" }) } + if (workflow.triggers !== undefined && workflow.triggers !== null && workflow.triggers.length > 0 && selectedTriggerIndex >= 0 && selectedTriggerIndex < workflow.triggers.length) { + console.log(workflow.triggers[selectedTriggerIndex]) + } + const UserinputSidebar = Object.getOwnPropertyNames(selectedTrigger).length === 0 || workflow.triggers[selectedTriggerIndex] === undefined || selectedTrigger.trigger_type !== "USERINPUT" ? null :

@@ -14994,7 +15050,12 @@ const releaseToConnectLabel = "Release to Connect" const newname = (option.name.charAt(0).toUpperCase() + option.name.substring(1)).replaceAll("_", " "); return newname; }} - options={workflows} + options={ + [{ + "id": "", + "name": "No Workflow Selected", + }].concat(workflows) + } fullWidth style={{ backgroundColor: theme.palette.inputColor, @@ -15149,14 +15210,40 @@ const releaseToConnectLabel = "Release to Connect"

-
+
Enabled Input-Questions {workflow.input_questions !== undefined && workflow.input_questions !== null && workflow.input_questions.length > 0 ?
{workflow.input_questions.map((question, index) => { + var foundParamIndex = workflow.triggers[selectedTriggerIndex].parameters.findIndex((param) => param.name === "input_questions") + const selectionClick = () => { - console.log("Clicked input question: ", question) - console.log("PARAMS: ", workflow.triggers[selectedTriggerIndex].parameters) + if (foundParamIndex === -1) { + workflow.triggers[selectedTriggerIndex].parameters.push({ + "name": "input_questions", + "value": [], + }) + + foundParamIndex = workflow.triggers[selectedTriggerIndex].parameters.length - 1 + } else { + try { + workflow.triggers[selectedTriggerIndex].parameters[foundParamIndex].value = JSON.parse(workflow.triggers[selectedTriggerIndex].parameters[foundParamIndex].value) + } catch (e) { + console.log("Couldn't parse input questions: ", e) + workflow.triggers[selectedTriggerIndex].parameters[foundParamIndex].value = [] + } + } + + if (workflow.triggers[selectedTriggerIndex].parameters[foundParamIndex].value.includes(question.name)) { + workflow.triggers[selectedTriggerIndex].parameters[foundParamIndex].value = workflow.triggers[selectedTriggerIndex].parameters[foundParamIndex].value.filter((item) => item !== question.name) + } else { + workflow.triggers[selectedTriggerIndex].parameters[foundParamIndex].value.push(question.name) + } + + // Make it back to a string + workflow.triggers[selectedTriggerIndex].parameters[foundParamIndex].value = JSON.stringify(workflow.triggers[selectedTriggerIndex].parameters[foundParamIndex].value) + setWorkflow(workflow) + setUpdate(Math.random()) } return ( @@ -15164,7 +15251,7 @@ const releaseToConnectLabel = "Release to Connect" selectionClick() }}> {question.name} @@ -15176,6 +15263,7 @@ const releaseToConnectLabel = "Release to Connect" :
{ setEditWorkflowModalOpen(true) + toast.info("Expand and scroll down to add input-questions") }}> No Input-Questions found. Click to add them!
diff --git a/frontend/src/views/Workflows.jsx b/frontend/src/views/Workflows.jsx index 0b642d56..816f17c5 100755 --- a/frontend/src/views/Workflows.jsx +++ b/frontend/src/views/Workflows.jsx @@ -1162,10 +1162,11 @@ const Workflows = (props) => { }) .then((responseJson) => { if (responseJson !== undefined) { + var newarray = [] - for (var key in responseJson) { - const wf = responseJson[key] - if (wf.public === true) { + for (var wfkey in responseJson) { + const wf = responseJson[wfkey] + if (wf.public === true || wf.hidden === true) { continue } @@ -1178,9 +1179,9 @@ const Workflows = (props) => { var parsedactionlist = []; for (var key in newarray) { const workflow = newarray[key] - if (workflow.status === "production") { - setProdFilter = true - } + //if (workflow.status === "production") { + // setProdFilter = true + //} for (var actionkey in newarray[key].actions) { const action = newarray[key].actions[actionkey]; From 52227397a9d0118b4a574b77dfbe1bfe9b4780c7 Mon Sep 17 00:00:00 2001 From: Frikky Date: Sat, 22 Jun 2024 23:44:01 +0200 Subject: [PATCH 5/7] Fixed user input to work better --- backend/go-app/walkoff.go | 2 +- frontend/src/components/EditWorkflow.jsx | 4 +- frontend/src/views/AngularWorkflow.jsx | 39 ++-- frontend/src/views/RunWorkflow.jsx | 245 ++++++++++++++++++----- 4 files changed, 211 insertions(+), 79 deletions(-) diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index 8ea1550a..7ff135e3 100755 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -1076,7 +1076,7 @@ func handleExecution(id string, workflow shuffle.Workflow, request *http.Request // return workflowExecution, fmt.Sprintf("%s", err), nil } else { log.Printf("[ERROR] Failed in prepareExecution: '%s'", err) - return shuffle.WorkflowExecution{}, fmt.Sprintf("Failed starting workflow: %s", err), err + return shuffle.WorkflowExecution{}, fmt.Sprintf("Failed running: %s", err), err } } diff --git a/frontend/src/components/EditWorkflow.jsx b/frontend/src/components/EditWorkflow.jsx index cc4a33b4..e2443664 100644 --- a/frontend/src/components/EditWorkflow.jsx +++ b/frontend/src/components/EditWorkflow.jsx @@ -685,8 +685,8 @@ const EditWorkflow = (props) => { Input fields - - Input fields are fields that will be used during the startup of the workflow. These will be formatted in JSON and is most commonly used from the workflow run page. + + Input fields are fields that will be used during the startup of the workflow. These will be formatted in JSON and is most commonly used from the workflow run page. If chosen in the User Input node, these will be required fields. diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index fb28083d..baab3dab 100755 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -1073,22 +1073,20 @@ const releaseToConnectLabel = "Release to Connect" // User Input & Subflow nodes if (param.name === "workflow" || param.name === "subflow") { - const paramIndex = param.name === "workflow" ? 0 : 5 - console.log("Current vs new: ", workflow.triggers[trigger_index].parameters[paramIndex].value, subworkflow.id) - - if (workflow.triggers[trigger_index].parameters[paramIndex].value !== subworkflow.id) { - if (param.value === workflow.id) { - setSubworkflow(workflow); - baseSubflow = workflow - } else { - const sub = responseJson.find((data) => data.id === param.value); - if (sub !== undefined && subworkflow.id !== sub.id) { - baseSubflow = sub - setSubworkflow(sub); - } - } - } - } + const paramIndex = param.name === "workflow" ? 0 : 5 + if (workflow.triggers[trigger_index].parameters[paramIndex].value !== subworkflow.id) { + if (param.value === workflow.id) { + setSubworkflow(workflow); + baseSubflow = workflow + } else { + const sub = responseJson.find((data) => data.id === param.value); + if (sub !== undefined && subworkflow.id !== sub.id) { + baseSubflow = sub + setSubworkflow(sub); + } + } + } + } if (param.name === "startnode" && param.value !== undefined && param.value !== null) { @@ -3255,7 +3253,6 @@ const releaseToConnectLabel = "Release to Connect" const getChildWorkflows = (parentWorkflowId) => { if (originalWorkflow.suborg_distribution === undefined || originalWorkflow.suborg_distribution === null || originalWorkflow.suborg_distribution.length === 0) { - console.log("No suborg distribution") return } @@ -14866,10 +14863,6 @@ const releaseToConnectLabel = "Release to Connect" }) } - if (workflow.triggers !== undefined && workflow.triggers !== null && workflow.triggers.length > 0 && selectedTriggerIndex >= 0 && selectedTriggerIndex < workflow.triggers.length) { - console.log(workflow.triggers[selectedTriggerIndex]) - } - const UserinputSidebar = Object.getOwnPropertyNames(selectedTrigger).length === 0 || workflow.triggers[selectedTriggerIndex] === undefined || selectedTrigger.trigger_type !== "USERINPUT" ? null :

@@ -15217,7 +15210,7 @@ const releaseToConnectLabel = "Release to Connect"

- Enabled Input-Questions + Required Input-Questions {workflow.input_questions !== undefined && workflow.input_questions !== null && workflow.input_questions.length > 0 ?
{workflow.input_questions.map((question, index) => { @@ -17156,10 +17149,12 @@ const releaseToConnectLabel = "Release to Connect" //defaultReturn = return null; } else { + /* console.log( "Unable to handle invalid trigger type " + selectedTrigger.trigger_type ); + */ return null; } } else if (Object.getOwnPropertyNames(selectedEdge).length > 0) { diff --git a/frontend/src/views/RunWorkflow.jsx b/frontend/src/views/RunWorkflow.jsx index ee1802d7..4d8dc1bf 100644 --- a/frontend/src/views/RunWorkflow.jsx +++ b/frontend/src/views/RunWorkflow.jsx @@ -2,6 +2,7 @@ import React, {useState, useEffect} from 'react'; import ReactDOM from "react-dom" +import { ToastContainer, toast } from "react-toastify" import { useInterval } from "react-powerhooks"; import { makeStyles } from '@mui/material/styles'; import { useNavigate, Link, useParams } from "react-router-dom"; @@ -43,7 +44,8 @@ const bodyDivStyle = { const RunWorkflow = (defaultprops) => { const { globalUrl, isLoaded, isLoggedIn, setIsLoggedIn, setCookie, register, serverside } = defaultprops; - let navigate = useNavigate(); + let navigate = useNavigate(); + const [_, setUpdate] = useState(""); // Used to force rendring, don't remove const [message, setMessage] = useState(""); const [workflow, setWorkflow] = React.useState({}); const [executionRequest, setExecutionRequest] = React.useState({}); @@ -55,6 +57,7 @@ const RunWorkflow = (defaultprops) => { const [selectedOrganization, setSelectedOrganization] = React.useState(undefined); const [apps, setApps] = React.useState([]); const [buttonClicked, setButtonClicked] = React.useState(""); + const [foundSourcenode, setFoundSourcenode] = React.useState(undefined); const boxStyle = { color: "white", @@ -340,6 +343,8 @@ const RunWorkflow = (defaultprops) => { event.preventDefault() } + console.log("ONSUBMIT: ", event, execution_id, authorization, answer) + stop() setMessage("") setExecutionLoading(true) @@ -389,12 +394,27 @@ const RunWorkflow = (defaultprops) => { url += `?reference_execution=${execution_id}&authorization=${authorization}&answer=${answer}` data = {} fetchBody.method = "GET" + + if (executionArgument !== undefined && executionArgument !== null) { + try { + if (typeof executionArgument === "string") { + url += "¬e=" + executionArgument + } else { + url += "¬e=" + JSON.stringify(executionArgument) + } + } catch (e) { + url += "¬e=" + executionArgument + } + } + } else { fetchBody.method = "POST" fetchBody.body = JSON.stringify(data) } - console.log("Pre request: ", url, fetchBody) + // IF there is an execution argument, we should use it + console.log("FULL URL: ", url) + fetch(url, fetchBody) .then((response) => { if (response.status !== 200 && response.status !== 201) { @@ -408,16 +428,26 @@ const RunWorkflow = (defaultprops) => { }) start(); - return + return response.json() } } - return response.json(); + return response.json() }) .then(responseJson => { setExecutionLoading(false) - if (responseJson["success"] === false) { + if (responseJson.success === false) { console.log("Failed sending execution request") + if (responseJson.reason !== undefined && responseJson.reason !== null) { + toast.warn(responseJson.reason) + } + + stop() + setMessage("") + setExecutionData({}) + setExecutionInfo("") + setExecutionRunning(false) + setExecutionRequest({}) } else { console.log("Started execution") @@ -426,17 +456,24 @@ const RunWorkflow = (defaultprops) => { } else { setExecutionRunning(true); setExecutionRequest(responseJson) - start(); + start() } } }) .catch(error => { //setExecutionInfo("Error in workflow startup: " + error) + toast.warn("Error in workflow startup: " + error) + + stop() + setMessage("") + setExecutionData({}) + setExecutionInfo("") + setExecutionLoading(false) }) } - const getWorkflow = (workflow_id) => { + const getWorkflow = (workflow_id, selectedNode) => { fetch(globalUrl + "/api/v1/workflows/" + workflow_id, { method: "GET", headers: { @@ -480,6 +517,48 @@ const RunWorkflow = (defaultprops) => { setExecutionArgument(newexec) } + if (selectedNode !== undefined && selectedNode !== null && selectedNode.length > 0) { + var found = false + for (var actionkey in responseJson.actions) { + if (responseJson.actions[actionkey].id === selectedNode) { + found = true + setFoundSourcenode(responseJson.actions[actionkey]) + break + } + } + + if (!found) { + for (var triggerkey in responseJson.triggers) { + if (responseJson.triggers[triggerkey].id !== selectedNode) { + continue + } + + setFoundSourcenode(responseJson.triggers[triggerkey]) + + if (responseJson.input_questions !== undefined && responseJson.input_questions !== null && responseJson.input_questions.length > 0 && responseJson.triggers[triggerkey].trigger_type === "USERINPUT") { + + // Look for input questions param + for (var paramkey in responseJson.triggers[triggerkey].parameters) { + if (responseJson.triggers[triggerkey].parameters[paramkey].name === "input_questions") { + + var relevantquestions = [] + for (var questionkey in responseJson.input_questions) { + if (responseJson.triggers[triggerkey].parameters[paramkey].value.includes(responseJson.input_questions[questionkey].name)) { + relevantquestions.push(responseJson.input_questions[questionkey]) + } + } + + responseJson.input_questions = relevantquestions + } + } + } + + + break + } + } + } + handleGetOrg(responseJson.org_id) setWorkflow(responseJson); }) @@ -489,7 +568,7 @@ const RunWorkflow = (defaultprops) => { }; const { start, stop } = useInterval({ - duration: 3000, + duration: 1500, startImmediate: true, callback: () => { fetchUpdates(executionRequest.execution_id, executionRequest.authorization) @@ -526,15 +605,11 @@ const RunWorkflow = (defaultprops) => { for (var key in responseJson.results) { if (responseJson.results[key].status === "WAITING") { - console.log("Found: ", responseJson.results[key]) - const validate = validateJson(responseJson.results[key].result) - console.log("Validate: ", validate) if (validate.valid && typeof validate.result === "string") { validate.result = JSON.parse(validate.result) } - console.log("Newresult: ", validate.result) if (validate.result["information"] !== undefined && validate.result["information"] !== null) { setWorkflowQuestion(validate.result["information"]) } @@ -623,15 +698,15 @@ const RunWorkflow = (defaultprops) => { return } - fetch(globalUrl + "/api/v1/streams/results", { - method: "POST", - headers: { - "Content-Type": "application/json", - Accept: "application/json", - }, - body: JSON.stringify(innerRequest), - credentials: "include", - }) + fetch(globalUrl + "/api/v1/streams/results", { + method: "POST", + headers: { + "Content-Type": "application/json", + Accept: "application/json", + }, + body: JSON.stringify(innerRequest), + credentials: "include", + }) .then((response) => { if (response.status !== 200) { console.log("Status not 200 for stream results :O!"); @@ -651,11 +726,14 @@ const RunWorkflow = (defaultprops) => { }); }; - const answer = new URLSearchParams(window.location.search).get("answer") - const execution_id = new URLSearchParams(window.location.search).get("reference_execution") - const authorization = new URLSearchParams(window.location.search).get("authorization") + const searchParams = new URLSearchParams(window.location.search) + const answer = searchParams.get("answer") + const execution_id = searchParams.get("reference_execution") + const authorization = searchParams.get("authorization") + const sourceNode = searchParams.get("source_node") + useEffect(() => { - getWorkflow(props.match.params.key) + getWorkflow(props.match.params.key, sourceNode) if (execution_id !== undefined && execution_id !== null && authorization !== undefined && authorization !== null) { console.log("Get execution: ", execution_id) fetchUpdates(execution_id, authorization, true) @@ -667,13 +745,60 @@ const RunWorkflow = (defaultprops) => { }, []) + useEffect(() => { + if (executionData === undefined || executionData === null || executionData === {}) { + return + } + + if (foundSourcenode === undefined || foundSourcenode === null || foundSourcenode === {}) { + return + } + + if (foundSourcenode.trigger_type !== "USERINPUT") { + return + } + + if (executionData.results === undefined || executionData.results === null || executionData.results.length === 0) { + return + } + + for (var resultkey in executionData.results) { + const result = executionData.results[resultkey] + if (result.action.id !== foundSourcenode.id) { + continue + } + + var parsedresult = result.result + try { + parsedresult = JSON.parse(parsedresult) + } catch (e) { + console.log("Error parsing result: ", e) + } + + if (result.status !== "WAITING") { + if (parsedresult.click_info !== undefined && parsedresult.click_info !== null) { + if (parsedresult.click_info.user !== undefined && parsedresult.click_info.user !== null && parsedresult.click_info.user.length > 0) { + setMessage("Already answered by " + parsedresult.click_info.user) + } + } else { + setMessage("Answered.") + } + + } + + break + } + + }, [executionData, foundSourcenode]) const buttonBackground = "linear-gradient(to right, #f86a3e, #f34079)" const buttonStyle = {borderRadius: 25, height: 50, fontSize: 18, backgroundImage: handleValidateForm(executionArgument) || executionLoading ? buttonBackground : "grey", color: "white"} - //console.log("execdata: ", executionData) const disabledButtons = message.length > 0 || executionData.status === "FINISHED" || executionData.status === "ABORTED" + //{disabledButtons ? null : + + const organization = selectedOrganization !== undefined && selectedOrganization !== null ? selectedOrganization.name : "Unknown" const contact = selectedOrganization !== undefined && selectedOrganization !== null && selectedOrganization.org !== undefined && selectedOrganization.org !== null? selectedOrganization.org : "support@shuffler.io" //const contact = selectedOrganization !== undefined && selectedOrganization !== null && selectedOrganization.contact !== undefined && selectedOrganization.contact !== null? selectedOrganization.contact : "support@shuffler.io" @@ -724,49 +849,54 @@ const RunWorkflow = (defaultprops) => { {organization} - - {contact} - - {message} + + {disabledButtons && message.length > 0 ? null : + + {message} + + } {answer !== undefined && answer !== null ? null : - {workflow.name} + {workflow.name} } {workflowQuestion.length > 0 ? - - {workflowQuestion} - +
+ + {workflowQuestion} + +
: null} {workflow.input_questions !== undefined && workflow.input_questions !== null && workflow.input_questions.length > 0 ?
{workflow.input_questions.map((question, index) => { - return (
{question.name} { + onBlur={(e) => { //setExecutionArgument(e.target.value) executionArgument[question.value] = e.target.value + setUpdate(Math.random()) }} />
@@ -813,20 +943,27 @@ const RunWorkflow = (defaultprops) => { : null} - : - answer !== undefined && answer !== null ? + ((answer !== undefined && answer !== null) || (foundSourcenode !== undefined && foundSourcenode !== null)) ? - - {disabledButtons ? "Already answered. Nothing to do." : ""} - + + {disabledButtons && message.length > 0 ? + + {message}. You may close this window. + + : + + {disabledButtons ? "Answered. You may close this window." : ""} + + } + {disabledButtons ? null : What do you want to do? }
-
} - {buttonClicked !== undefined && buttonClicked !== null && buttonClicked !== "finished" && buttonClicked.length > 0 ? + {/*buttonClicked !== undefined && buttonClicked !== null && buttonClicked !== "finished" && buttonClicked.length > 0 ? finalize workflow animation { console.log("Img loaded.") @@ -868,7 +1005,7 @@ const RunWorkflow = (defaultprops) => { }} /> - : null} + : null*/}
{executionInfo} From e1f7a1e9d35811da8de63b082a516eaf26a7a1ec Mon Sep 17 00:00:00 2001 From: Frikky Date: Mon, 24 Jun 2024 15:28:21 +0200 Subject: [PATCH 6/7] Updated some workflow issues --- backend/go-app/go.mod | 2 +- frontend/src/components/Billing.jsx | 4 +- frontend/src/components/NewHeader.jsx | 3 +- frontend/src/components/ParsedAction.jsx | 30 ++++- frontend/src/views/Admin.jsx | 10 +- frontend/src/views/AngularWorkflow.jsx | 155 ++++++++++++----------- 6 files changed, 119 insertions(+), 85 deletions(-) diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod index ebe85b01..84fcf764 100644 --- a/backend/go-app/go.mod +++ b/backend/go-app/go.mod @@ -20,7 +20,7 @@ require ( github.com/gorilla/mux v1.8.1 github.com/h2non/filetype v1.1.3 github.com/satori/go.uuid v1.2.0 - github.com/shuffle/shuffle-shared v0.6.47 + github.com/shuffle/shuffle-shared v0.6.50 golang.org/x/crypto v0.22.0 google.golang.org/api v0.176.1 google.golang.org/grpc v1.63.2 diff --git a/frontend/src/components/Billing.jsx b/frontend/src/components/Billing.jsx index e567a5c7..cb18d20d 100644 --- a/frontend/src/components/Billing.jsx +++ b/frontend/src/components/Billing.jsx @@ -1775,7 +1775,7 @@ const Billing = (props) => {