Merge branch 'main' of https://github.com/shuffle/shuffle-shared

Author: frikky

db-connector.go

@@ -1583,6 +1583,9 @@ func GetExecutionVariables(ctx context.Context, executionId string) (string, int
 }
 
 func getExecutionFileValue(ctx context.Context, workflowExecution WorkflowExecution, action ActionResult) (string, error) {
+	projectName := os.Getenv("SHUFFLE_GCEPROJECT")
+	bucketName := project.BucketName
+
 	fullParsedPath := fmt.Sprintf("large_executions/%s/%s_%s", workflowExecution.ExecutionOrg, workflowExecution.ExecutionId, action.Action.ID)
 
 	cacheKey := fmt.Sprintf("%s_%s_action_replace", workflowExecution.ExecutionId, action.Action.ID)
@@ -1594,11 +1597,24 @@ func getExecutionFileValue(ctx context.Context, workflowExecution WorkflowExecut
 		}
 	}
 
-	bucket := project.StorageClient.Bucket(project.BucketName)
+	bucket := project.StorageClient.Bucket(bucketName)
 	obj := bucket.Object(fullParsedPath)
 	fileReader, err := obj.NewReader(ctx)
 	if err != nil {
-		return "", err
+		log.Printf("[ERROR] Failed reading file from bucket %s: %s. Will try with alternative solution.", bucketName, err)
+
+		if projectName != "shuffler" {
+			bucketName = fmt.Sprintf("%s.appspot.com", projectName)
+			bucket = project.StorageClient.Bucket(bucketName)
+			obj = bucket.Object(fullParsedPath)
+			fileReader, err = obj.NewReader(ctx)
+			if err != nil {
+				log.Printf("[ERROR] Failed reading file again from bucket %s: %s", bucketName, err)
+				return "", err
+			}
+		} else {
+			return "", err
+		}
 	}
 
 	data, err := ioutil.ReadAll(fileReader)

health.go

@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"log"
 	"net/http"
@@ -36,6 +37,11 @@ type executionResult struct {
 	ID      string `json:"id"`
 }
 
+type testRun struct {
+	CRUrl string `json:"cloudrun_url"`
+	Region string `json:"region"`
+}
+
 func updateOpsCache(workflowHealth WorkflowHealth) {
 	cacheKey := fmt.Sprintf("ops-health-check")
 	ctx := context.Background()
@@ -587,7 +593,7 @@ func RunOpsHealthCheck(resp http.ResponseWriter, request *http.Request) {
 	errorChannel := make(chan error)
 	go func() {
 		log.Printf("[DEBUG] Running workflowHealthChannel goroutine")
-		workflowHealth, err := RunOpsWorkflow(apiKey, orgId)
+		workflowHealth, err := RunOpsWorkflow(apiKey, orgId, "")
 		if err != nil {
 			log.Printf("[ERROR] Failed workflow health check: %s", err)
 		}
@@ -887,7 +893,7 @@ func fixOpensearch() error {
 	return nil
 }
 
-func RunOpsWorkflow(apiKey string, orgId string) (WorkflowHealth, error) {
+func RunOpsWorkflow(apiKey string, orgId string, cloudRunUrl string) (WorkflowHealth, error) {
 	// run workflow with id 602c7cf5-500e-4bd1-8a97-aa5bc8a554e6
 	ctx := context.Background()
 
@@ -902,11 +908,15 @@ func RunOpsWorkflow(apiKey string, orgId string) (WorkflowHealth, error) {
 	}
 
 	baseUrl := os.Getenv("SHUFFLE_CLOUDRUN_URL")
-	if len(baseUrl) == 0 {
+	if len(baseUrl) == 0 && (cloudRunUrl == "" || len(cloudRunUrl) == 0) {
 		log.Printf("[DEBUG] Base url not set. Setting to default")
 		baseUrl = "https://shuffler.io"
 	}
 
+	if len(baseUrl) == 0 {
+		baseUrl = cloudRunUrl
+	}
+
 	if project.Environment == "onprem" {
 		log.Printf("[DEBUG] Onprem environment. Setting base url to localhost")
 		baseUrl = "http://localhost:5001"
@@ -1081,6 +1091,19 @@ func RunOpsWorkflow(apiKey string, orgId string) (WorkflowHealth, error) {
 			workflowHealth.RunStatus = executionResults.Status
 		}
 
+		if executionResults.Status == "FINISHED" {
+			log.Printf("[DEBUG] Workflow Health exeution is finished, checking it's results")
+
+			// yash asked to comment these out
+			// for _, r := range executionResults.Results {
+			// 	if r.Status != "SUCCESS" {
+			// 		workflowHealth.RunStatus = "FAILED"
+			// 		break
+			// 	}
+			// }
+		}
+
+
 		updateOpsCache(workflowHealth)
 
 		//log.Printf("[DEBUG] Workflow Health execution Result Status: %#v for executionID: %s", executionResults.Status, workflowHealth.ExecutionId)
@@ -1108,6 +1131,38 @@ func RunOpsWorkflow(apiKey string, orgId string) (WorkflowHealth, error) {
 	return workflowHealth, nil
 }
 
+func RunHealthTest(resp http.ResponseWriter, req *http.Request) {
+	response, err := io.ReadAll(req.Body)
+	if err != nil {
+		log.Printf("[ERROR] Failed to read body of the health test case: %s", err)
+		resp.WriteHeader(500)
+		resp.Write([]byte(`{"success": false, "reason": "failed to read the body"}`))
+		return
+	}
+
+	var execData testRun
+	err = json.Unmarshal(response, &execData)
+	if err != nil {
+		log.Printf("[ERROR] Error unmarshaling test data: %s", err)
+		resp.WriteHeader(500)
+		resp.Write([]byte(`{"success": false, "reason": "failed to unmarshal the data"}`))
+		return
+	}
+
+	apiKey := os.Getenv("SHUFFLE_OPS_DASHBOARD_APIKEY")
+	orgId := os.Getenv("SHUFFLE_OPS_DASHBOARD_ORG")
+
+	
+	health, err := RunOpsWorkflow(apiKey, orgId, execData.CRUrl)
+	if err != nil {
+		log.Printf("[ERROR] Health test failed %v", err)
+	}
+
+	jsonHealth, err := json.Marshal(health)
+	resp.WriteHeader(200)
+	resp.Write(jsonHealth)
+}
+
 func InitOpsWorkflow(apiKey string, OrgId string) (string, error) {
 	opsDashboardApikey := apiKey
 	opsDashboardOrgId := OrgId

shared.go

@@ -7748,6 +7748,7 @@ func SaveWorkflow(resp http.ResponseWriter, request *http.Request) {
 			log.Printf("[AUDIT] Letting verified support admin %s access workflow %s (save workflow)", user.Username, workflow.ID)
 
 			workflow.ID = tmpworkflow.ID
+
 		} else if tmpworkflow.OrgId == user.ActiveOrg.Id && user.Role != "org-reader" {
 			log.Printf("[AUDIT] User %s is accessing workflow %s (save workflow)", user.Username, tmpworkflow.ID)
 			workflow.ID = tmpworkflow.ID