From 074709c7bb6cccd8f570b6505a8d69d20a08db05 Mon Sep 17 00:00:00 2001 From: yashsinghcodes Date: Thu, 26 Jun 2025 20:02:50 +0530 Subject: [PATCH 1/7] fix orborus issues --- backend/go-app/walkoff.go | 24 ++++++++++++++-------- functions/onprem/orborus/orborus.go | 31 +++++++++++++++-------------- 2 files changed, 32 insertions(+), 23 deletions(-) diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index c194d091..0aff66c8 100755 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -246,16 +246,16 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) { } // This is really the environment's name - NOT org-id - orgId := request.Header.Get("Org-Id") - if len(orgId) == 0 { + environment := request.Header.Get("Org-Id") + if len(environment) == 0 { log.Printf("[AUDIT] No org-id header set") resp.WriteHeader(401) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Specify the org-id header."}`))) return } - environment := request.Header.Get("org") - if len(environment) == 0 { + orgId := request.Header.Get("org") + if len(orgId) == 0 { //log.Printf("[AUDIT] No 'org' header set (get workflow queue). ") /* resp.WriteHeader(403) @@ -280,9 +280,17 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) { //log.Printf("[AUDIT] Get workflow queue for org %s, env %s, orborus label %s", orgId, environment, orborusLabel) ctx := shuffle.GetContext(request) - env, err := shuffle.GetEnvironment(ctx, orgId, "") - if err != nil { - log.Printf("[WARNING] No env found matching %s - continuing without updating orborus anyway: %s", orgId, err) + envs, err := shuffle.GetEnvironments(ctx, orgId) + if err != nil || len(envs) == 0 { + log.Printf("[WARNING] No env found matching %s - continuing without updating orborus anyway: %s", environment, err) + } + + var env *shuffle.Environment + for _, e := range envs { + if e.Name == environment { + env = &e + break + } } timeNow := time.Now().Unix() @@ -450,7 +458,7 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) { } } - executionRequests, err := shuffle.GetWorkflowQueue(ctx, orgId, 100) + executionRequests, err := shuffle.GetWorkflowQueue(ctx, environment, 100) if err != nil { // Skipping as this comes up over and over //log.Printf("(2) Failed reading body for workflowqueue: %s", err) diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index 9b5c87da..e350082b 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -813,21 +813,21 @@ func handleBackendImageDownload(ctx context.Context, images string) error { newImages = append(newImages, curimage) // Force remove the current image to avoid cached layers - if swarmConfig == "run" || swarmConfig == "swarm" { - _, err := dockercli.ImageRemove(ctx, curimage, image.RemoveOptions{ - Force: true, - PruneChildren: true, - }) - - if err != nil { - log.Printf("[ERROR] Failed removing image for re-download: %s", err) - } else { - log.Printf("[DEBUG] Removed image: %s", curimage) - } - } else { - //log.Printf("[DEBUG] Skipping image removal for %s as swarmConfig is not set to run or swarm. Value: %#v", curimage, swarmConfig) - } - +// if swarmConfig == "run" || swarmConfig == "swarm" { +// _, err := dockercli.ImageRemove(ctx, curimage, image.RemoveOptions{ +// Force: true, +// PruneChildren: true, +// }) +// +// if err != nil { +// log.Printf("[ERROR] Failed removing image for re-download: %s", err) +// } else { +// log.Printf("[DEBUG] Removed image: %s", curimage) +// } +// } else { +// //log.Printf("[DEBUG] Skipping image removal for %s as swarmConfig is not set to run or swarm. Value: %#v", curimage, swarmConfig) +// } +// err := shuffle.DownloadDockerImageBackend(&http.Client{Timeout: imagedownloadTimeout}, curimage) if err != nil { //log.Printf("[ERROR] Failed downloading image: %s", err) @@ -874,6 +874,7 @@ func handleBackendImageDownload(ctx context.Context, images string) error { // Update the service to run with the new image //docker service update --image username/imagename:latest servicename --force serviceUpdateOptions := types.ServiceUpdateOptions{} + service.Spec.TaskTemplate.ForceUpdate++ resp, err := dockercli.ServiceUpdate( ctx, service.ID, From 9bb578616fd4407ddb159d8ffa4b98a0d63c4093 Mon Sep 17 00:00:00 2001 From: yashsinghcodes Date: Sat, 28 Jun 2025 17:26:48 +0530 Subject: [PATCH 2/7] checks for nil pointer --- backend/go-app/walkoff.go | 178 +++++--------------------------------- 1 file changed, 20 insertions(+), 158 deletions(-) diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index 0aff66c8..08825a52 100755 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -11,7 +11,7 @@ import ( "io" "io/ioutil" "log" - "math/rand" + //"math/rand" "net/http" "net/url" "os" @@ -265,7 +265,8 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) { } orborusLabel := request.Header.Get("x-orborus-label") - + _ = orborusLabel + // This section is cloud custom for now auth := request.Header.Get("Authorization") if len(auth) == 0 { @@ -286,13 +287,22 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) { } var env *shuffle.Environment - for _, e := range envs { - if e.Name == environment { - env = &e + found := false + for i := range envs { + if envs[i].Name == environment { + env = &envs[i] + found = true break } } + if !found { + log.Printf("[ERROR] Failed to find environment(%s) for org(%s)", environment, orgId) + resp.WriteHeader(400) + resp.Write([]byte(`{"success":false,"reason":"environment not found"}`)) + return + } + timeNow := time.Now().Unix() err = shuffle.HandleOrborusFailover(ctx, request, resp, env) if err != nil { @@ -300,162 +310,14 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) { log.Printf("[WARNING] Failed handling Orborus failover: %s", err) } - log.Printf("[DEBUG] Issue with environment ID: %s", orgId) + log.Printf("[DEBUG] Issue with environment ID: %s", environment) return } //log.Printf("Found env: %#v", env) if len(env.OrgId) > 0 { - environment = env.OrgId - } - - // FIXME: Workflow stats disabled for now - // as it caused too many problems - // goal: track docker stuff once a minute and graph it - // For now: Disable this as it caused too many problems - if request.Method == "POST" && true == false { - //log.Printf("[DEBUG] POST to workflowqueue") - if rand.Intn(10) == 0 { - // Parse out body - body, err := ioutil.ReadAll(request.Body) - if err == nil { - - // Parse out CPU, memory and disk. - - var envData shuffle.OrborusStats - err = json.Unmarshal(body, &envData) - if err == nil && !envData.Swarm && !envData.Kubernetes && (envData.CPU > 0 || envData.Memory > 0 || envData.Disk > 0) { - - // Set the input in memory - envData.OrgId = orgId - envData.Environment = environment - envData.OrborusLabel = orborusLabel - envData.Timestamp = time.Now().Unix() - - if envData.CPU > 0 && envData.MaxCPU > 0 { - envData.CPUPercent = float64(envData.CPU) / float64(envData.MaxCPU) - } - - if envData.Memory > 0 && envData.MaxMemory > 0 { - envData.MemoryPercent = float64(envData.Memory) / float64(envData.MaxMemory) - } - - // Check if CPU percent constantly has stayed above X% for the last Y requests - percentageCheck := 90 - concurrentChecks := 2 - - //if int(envData.CPUPercent) > percentageCheck { - // Get cached data - percentages := []float64{} - cacheKey := fmt.Sprintf("%s_%s_percent", orgId, strings.ToLower(environment)) - - // Marshal float list into []byte - cacheData := []byte{} - cache, err := shuffle.GetCache(ctx, cacheKey) - if err == nil { - // Unmarshal into percentages - cacheData := []byte(cache.([]uint8)) - err = json.Unmarshal(cacheData, &percentages) - if err != nil { - log.Printf("[INFO] error in cache unmarshal for percentages: %s", err) - } - - if len(percentages) > concurrentChecks { - percentages = percentages[:concurrentChecks] - } - - percentages = append(percentages, envData.CPUPercent) - if len(percentages) > concurrentChecks { - //log.Printf("[INFO] Checking percentages: %v", percentages) - - // percentageCheck := 1 - sendAlert := true - for _, p := range percentages { - if int(p) < percentageCheck { - //log.Printf("[AUDIT] CPU percent is below %d: %d", percentageCheck, int(p)) - sendAlert = false - break - } - } - - if sendAlert { - log.Printf("[INFO] CPU percent has been above %d percent for the last 5 requests. Sending alert. Env: %s, org: %s", percentageCheck, environment, orgId) - - // Set notification + alert for organization - err = shuffle.CreateOrgNotification( - ctx, - fmt.Sprintf("CPU percent has been above %d percent", percentageCheck), - fmt.Sprintf("A environment %s has been using more than %d\\% CPU for the last 5 requests.", environment, percentageCheck), - fmt.Sprintf("/admin?tab=environments"), - environment, - true, - ) - - if err != nil { - log.Printf("[ERROR] error creating notification: %s", err) - } - - org, err := shuffle.GetOrg(ctx, environment) - if err == nil { - foundRecommendation := false - for _, recommendation := range org.Priorities { - if strings.Contains(recommendation.Name, "CPU") { - foundRecommendation = true - break - } - } - - if !foundRecommendation { - // Add to start of org.Priorities - org, _ = shuffle.AddPriority(*org, shuffle.Priority{ - Name: fmt.Sprintf("High CPU in environment %s", orgId), - Description: fmt.Sprintf("The environment %s has been using more than %d percent CPU. This indicates you may need to look at scaling.", orgId, percentageCheck), - Type: "scale", - Active: true, - URL: fmt.Sprintf("/admin?tab=environments"), - Severity: 1, - }, false) - - //Make last item the first item - org.Priorities = append([]shuffle.Priority{org.Priorities[len(org.Priorities)-1]}, org.Priorities[:len(org.Priorities)-1]...) - err = shuffle.SetOrg(ctx, *org, org.Id) - if err != nil { - log.Printf("[ERROR] Problem setting org: %s", err) - } - } - } - } - - if len(percentages) > 1 { - percentages = percentages[1:] - } - } - - // Marshal float list into []byte - } else { - //log.Printf("[ERROR] Failed getting cache: %s", err) - percentages = append(percentages, envData.CPUPercent) - } - - if len(percentages) > 0 { - //log.Printf("[DEBUG] Setting cache for %s: %#v", cacheKey, percentages) - cacheData, err = json.Marshal(percentages) - if err != nil { - log.Printf("[INFO] error in cache marshal: %s", err) - } - - // Add the new data - go shuffle.SetCache(ctx, cacheKey, cacheData, 5) - } - } - - //log.Printf("CPU percent: %f", envData.CPUPercent) - //log.Printf("Memory percent: %f", envData.MemoryPercent*100) - - go shuffle.SetenvStats(ctx, envData) - } - } + orgId = env.OrgId } executionRequests, err := shuffle.GetWorkflowQueue(ctx, environment, 100) @@ -486,10 +348,10 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) { } } - if len(orgId) > 0 { - env, err := shuffle.GetEnvironment(ctx, orgId, foundId) + if len(environment) > 0 { + env, err := shuffle.GetEnvironment(ctx, environment, foundId) if err != nil { - log.Printf("[WARNING] No env found matching %s - continuing without updating orborus anyway: %s", orgId, err) + log.Printf("[WARNING] No env found matching %s - continuing without updating orborus anyway: %s", environment, err) //resp.WriteHeader(401) //resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "No env found matching %s"}`, id))) //return From 0be2b9fe93cb9d4e8277dc7771bc3c55a9fe7092 Mon Sep 17 00:00:00 2001 From: yashsinghcodes Date: Sat, 28 Jun 2025 19:18:52 +0530 Subject: [PATCH 3/7] log org from the header --- backend/go-app/walkoff.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index 08825a52..0f078dca 100755 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -254,9 +254,9 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) { return } - orgId := request.Header.Get("org") + orgId := request.Header.Get("Org") if len(orgId) == 0 { - //log.Printf("[AUDIT] No 'org' header set (get workflow queue). ") + log.Printf("[AUDIT] No 'org' header set (get workflow queue). ") /* resp.WriteHeader(403) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Specify the org header. This can be done by setting the 'ORG' environment variable for Orborus to your Org ID in Shuffle"}`))) From 232b0263c661d164566acf4a480091cd239995cf Mon Sep 17 00:00:00 2001 From: Aditya <60684641+0x0elliot@users.noreply.github.com> Date: Wed, 2 Jul 2025 20:19:56 +0530 Subject: [PATCH 4/7] fix: enabling /api/v1/orgs for prod --- backend/go-app/main.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/backend/go-app/main.go b/backend/go-app/main.go index e1f0defa..06412fb4 100755 --- a/backend/go-app/main.go +++ b/backend/go-app/main.go @@ -5221,8 +5221,8 @@ func initHandlers() { // EVERYTHING below here is NEW for 0.8.0 (written 25.05.2021) r.HandleFunc("/api/v1/workflows/{key}/publish", makeWorkflowPublic).Methods("POST", "OPTIONS") r.HandleFunc("/api/v1/cloud/setup", handleCloudSetup).Methods("POST", "OPTIONS") - //r.HandleFunc("/api/v1/orgs", shuffle.HandleGetOrgs).Methods("GET", "OPTIONS") - //r.HandleFunc("/api/v1/orgs/", shuffle.HandleGetOrgs).Methods("GET", "OPTIONS") + r.HandleFunc("/api/v1/orgs", shuffle.HandleGetOrgs).Methods("GET", "OPTIONS") + r.HandleFunc("/api/v1/orgs/", shuffle.HandleGetOrgs).Methods("GET", "OPTIONS") r.HandleFunc("/api/v1/orgs/{orgId}", shuffle.HandleGetOrg).Methods("GET", "OPTIONS") r.HandleFunc("/api/v1/orgs/{orgId}", shuffle.HandleEditOrg).Methods("POST", "OPTIONS") r.HandleFunc("/api/v1/orgs/{orgid}/forms", shuffle.HandleGetOrgForms).Methods("GET", "OPTIONS") From aa1c98240bdae0d0030e0caf78158f7e676bb0bf Mon Sep 17 00:00:00 2001 From: Aditya <60684641+0x0elliot@users.noreply.github.com> Date: Wed, 2 Jul 2025 20:59:49 +0530 Subject: [PATCH 5/7] ci: adding tagged nightly release back --- .github/workflows/tagged-nightly-release.yaml | 92 +++++++++++++++++++ 1 file changed, 92 insertions(+) create mode 100644 .github/workflows/tagged-nightly-release.yaml diff --git a/.github/workflows/tagged-nightly-release.yaml b/.github/workflows/tagged-nightly-release.yaml new file mode 100644 index 00000000..c24b7e1d --- /dev/null +++ b/.github/workflows/tagged-nightly-release.yaml @@ -0,0 +1,92 @@ +name: Tagged Nightly Release +on: + release: + types: [published] + branches: + - nightly + +# there is a very clear point to this existing. +# we want to also release versions that look like this: +# v2.1.0-nightly-date, v2.1.0-nightly-date-1, v2.1.0-nightly-date-2 +# we NEVER want to send customers a "nightly" tag. We always want to send them +# a tagged nightly tag. So that when something breaks, They can always +# point to it. + +jobs: + main: + runs-on: ubuntu-latest + continue-on-error: ${{ matrix.experimental }} + strategy: + fail-fast: false + matrix: + include: + - app: frontend + path: frontend + experimental: true + - app: backend + path: backend + experimental: true + - app: app_sdk + path: backend/app_sdk + experimental: true + - app: orborus + path: functions/onprem/orborus + experimental: true + - app: worker + path: functions/onprem/worker + experimental: true + steps: + - name: Checkout + uses: actions/checkout@v3 + + - name: Set version + id: set_version + run: | + if [[ ${{ github.event_name }} == 'release' ]]; then + echo "VERSION=${{ github.event.release.tag_name }}" >> $GITHUB_OUTPUT + else + echo "VERSION=nightly-untagged-latest" >> $GITHUB_OUTPUT + fi + + - name: Set up Docker Buildx + uses: docker/setup-buildx-action@v3 + + - name: Set up QEMU + uses: docker/setup-qemu-action@v3 + with: + platforms: "amd64,arm64,arm" + + - name: Login to DockerHub + uses: docker/login-action@v3 + with: + username: ${{ secrets.DOCKERHUB_USERNAME }} + password: ${{ secrets.DOCKERHUB_TOKEN }} + + - name: Login to Ghcr + uses: docker/login-action@v3 + with: + registry: ghcr.io + username: ${{ github.actor }} + password: ${{ secrets.GITHUB_TOKEN }} + + - name: Ghcr Build and push + id: docker_build + uses: docker/build-push-action@v4 + env: + BUILDX_NO_DEFAULT_LOAD: true + with: + logout: false + context: ${{ matrix.path }}/ + file: ${{ matrix.path }}/Dockerfile + platforms: linux/amd64,linux/arm64 + push: true + cache-from: type=local,src=/tmp/.buildx-cache + cache-to: type=local,dest=/tmp/.buildx-cache + tags: | + ghcr.io/shuffle/shuffle-${{ matrix.app }}:${{ steps.set_version.outputs.VERSION }} + ${{ secrets.DOCKERHUB_USERNAME }}/shuffle-${{ matrix.app }}:${{ steps.set_version.outputs.VERSION }} + frikky/shuffle-${{ matrix.app }}:${{ steps.set_version.outputs.VERSION }} + frikky/shuffle:${{ matrix.app }} + + - name: Image digest + run: echo ${{ steps.docker_build.outputs.digest }} \ No newline at end of file From 48b18635d2d1a4b8d25083de435791879b599616 Mon Sep 17 00:00:00 2001 From: yashsinghcodes Date: Wed, 9 Jul 2025 14:03:16 +0530 Subject: [PATCH 6/7] removed return onprem --- backend/go-app/walkoff.go | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index 0f078dca..172cc0b4 100755 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -306,13 +306,7 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) { timeNow := time.Now().Unix() err = shuffle.HandleOrborusFailover(ctx, request, resp, env) if err != nil { - if !strings.Contains(err.Error(), "mismatch") { - log.Printf("[WARNING] Failed handling Orborus failover: %s", err) - } - - log.Printf("[DEBUG] Issue with environment ID: %s", environment) - - return + log.Printf("[WARNING] Failed handling Orborus failover: %s", err) } //log.Printf("Found env: %#v", env) From 98f5eabaaaafaa68ccaa0a7380445ada20daff21 Mon Sep 17 00:00:00 2001 From: yashsinghcodes Date: Wed, 9 Jul 2025 14:04:39 +0530 Subject: [PATCH 7/7] removed a spam log message --- backend/go-app/walkoff.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/backend/go-app/walkoff.go b/backend/go-app/walkoff.go index 172cc0b4..f2d8a118 100755 --- a/backend/go-app/walkoff.go +++ b/backend/go-app/walkoff.go @@ -256,7 +256,7 @@ func handleGetWorkflowqueue(resp http.ResponseWriter, request *http.Request) { orgId := request.Header.Get("Org") if len(orgId) == 0 { - log.Printf("[AUDIT] No 'org' header set (get workflow queue). ") + //log.Printf("[AUDIT] No 'org' header set (get workflow queue). ") /* resp.WriteHeader(403) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "Specify the org header. This can be done by setting the 'ORG' environment variable for Orborus to your Org ID in Shuffle"}`)))