diff --git a/.env b/.env index 64e325bc..298128cc 100755 --- a/.env +++ b/.env @@ -65,27 +65,33 @@ IS_KUBERNETES=false #SHUFFLE_BASE_IMAGE_NAME=shuffle #SHUFFLE_BASE_IMAGE_REGISTRY=ghcr.io -#SHUFFLE_BASE_IMAGE_TAG_SUFFIX="-1.3.1" +SHUFFLE_BASE_IMAGE_REPOSITORY=frikky +#SHUFFLE_BASE_IMAGE_TAG_SUFFIX="-1.4.0" # The eth0 interface inside a container corresponds # to the virtual Ethernet interface that connects # the container to the docker0 -SHUFFLE_SWARM_BRIDGE_DEFAULT_INTERFACE=eth0 -SHUFFLE_SWARM_BRIDGE_DEFAULT_MTU=1500 # 1500 by default +SHUFFLE_SWARM_BRIDGE_DEFAULT_INTERFACE=eth0 +# 1500 by default +SHUFFLE_SWARM_BRIDGE_DEFAULT_MTU=1500 # Used for auto-cleanup of containers. REALLY important at scale. Set to false to see all container info. SHUFFLE_MEMCACHED= SHUFFLE_CONTAINER_AUTO_CLEANUP=true -SHUFFLE_ORBORUS_EXECUTION_CONCURRENCY=5 # The amount of concurrent executions Orborus can handle. This is a soft limit, but it's recommended to keep it low. +# The amount of concurrent executions Orborus can handle. This is a soft limit, but it's recommended to keep it low. +SHUFFLE_ORBORUS_EXECUTION_CONCURRENCY=5 SHUFFLE_HEALTHCHECK_DISABLED=false SHUFFLE_ELASTIC=true SHUFFLE_LOGS_DISABLED=false -SHUFFLE_CHAT_DISABLED=false # Controls support chat +SHUFFLE_CHAT_DISABLED=false SHUFFLE_DISABLE_RERUN_AND_ABORT=false SHUFFLE_RERUN_SCHEDULE=300 -SHUFFLE_WORKER_SERVER_URL= # Definition in case Worker & Orborus is talking to the wrong server -SHUFFLE_ORBORUS_PULL_TIME= # Definition in case Orborus is pulling too often/not often enough -SHUFFLE_MAX_EXECUTION_DEPTH= # Max recursion depth for subflows +# Definition in case Worker & Orborus is talking to the wrong server +SHUFFLE_WORKER_SERVER_URL= +# Definition in case Orborus is pulling too often/not often enough +SHUFFLE_ORBORUS_PULL_TIME= +# Max recursion depth for subflows +SHUFFLE_MAX_EXECUTION_DEPTH= # DATABASE CONFIGURATIONS DATASTORE_EMULATOR_HOST=shuffle-database:8000 diff --git a/.github/workflows/dockerbuild.yaml b/.github/workflows/dockerbuild.yaml index 8f139bba..a14dfaa9 100644 --- a/.github/workflows/dockerbuild.yaml +++ b/.github/workflows/dockerbuild.yaml @@ -19,23 +19,23 @@ jobs: include: - app: frontend path: frontend - version: nightly + version: 1.4.0 experimental: true - app: backend path: backend - version: nightly + version: 1.4.0 experimental: true - app: app_sdk path: backend/app_sdk - version: nightly + version: 1.4.0 experimental: true - app: orborus path: functions/onprem/orborus - version: nightly + version: 1.4.0 experimental: true - app: worker path: functions/onprem/worker - version: nightly + version: 1.4.0 experimental: true steps: - name: Checkout @@ -77,11 +77,11 @@ jobs: cache-to: type=local,dest=/tmp/.buildx-cache tags: | ghcr.io/shuffle/shuffle-${{ matrix.app }}:${{ matrix.version }} - ghcr.io/shuffle/shuffle-${{ matrix.app }}:nightly + ghcr.io/shuffle/shuffle-${{ matrix.app }}:latest ${{ secrets.DOCKERHUB_USERNAME }}/shuffle-${{ matrix.app }}:${{ matrix.version }} - ${{ secrets.DOCKERHUB_USERNAME }}/shuffle-${{ matrix.app }}:nightly + ${{ secrets.DOCKERHUB_USERNAME }}/shuffle-${{ matrix.app }}:latest frikky/shuffle-${{ matrix.app }}:${{ matrix.version }} - frikky/shuffle-${{ matrix.app }}:nightly + frikky/shuffle-${{ matrix.app }}:latest frikky/shuffle:${{ matrix.app }} - name: Image digest diff --git a/README.md b/README.md index e8c2f9c1..3e22c1d4 100755 --- a/README.md +++ b/README.md @@ -15,16 +15,16 @@ Shuffle Automation [_Community & Support_](https://discord.gg/B2CBzUm) — [_Documentation_](https://shuffler.io/docs) — [_Getting Started_](https://shuffler.io/docs/getting_started) — -[_Development_](https://github.com/frikky/Shuffle/blob/master/.github/CONTRIBUTING.md) +[_Development_](https://github.com/shuffle/Shuffle/blob/master/.github/CONTRIBUTING.md) Follow us on Twitter at [@shuffleio](https://twitter.com/shuffleio). -![Example Shuffle webhook integration](https://github.com/frikky/Shuffle/blob/main/frontend/src/assets/img/github_shuffle_img.png) +![Example Shuffle webhook integration](https://github.com/shuffle/Shuffle/blob/main/frontend/src/assets/img/github_shuffle_img.png) ## Try it -* Self-hosted: Check out the [installation guide](https://github.com/frikky/shuffle/blob/master/.github/install-guide.md) +* Self-hosted: Check out the [installation guide](https://github.com/shuffle/shuffle/blob/master/.github/install-guide.md) * Cloud: Register at https://shuffler.io/register and get cooking (missing a lot of features) Please consider [sponsoring](https://github.com/sponsors/frikky) the project if you want to see more rapid development. @@ -33,7 +33,7 @@ Please consider [sponsoring](https://github.com/sponsors/frikky) the project if * [Discord](https://discord.gg/B2CBzUm) * [Twitter](https://twitter.com/shuffleio) * [Email](mailto:frikky@shuffler.io) -* [Open issue](https://github.com/frikky/Shuffle/issues/new) +* [Open issue](https://github.com/shuffle/Shuffle/issues/new) * [Shuffler.io](https://shuffler.io/contact) ## Blogposts @@ -43,21 +43,21 @@ Please consider [sponsoring](https://github.com/sponsors/frikky) the project if * [4. Real-time executions with TheHive, Cortex and MISP](https://medium.com/@Frikkylikeme/indicators-and-webhooks-with-thehive-cortex-and-misp-open-source-soar-part-4-f70cde942e59) ## Documentation -[Documentation](https://shuffler.io/docs) can be found on [https://shuffler.io/docs](https://shuffler.io/docs) and is written here: [https://github.com/frikky/shuffle-docs](https://github.com/frikky/shuffle-docs). +[Documentation](https://shuffler.io/docs) can be found on [https://shuffler.io/docs](https://shuffler.io/docs) and is written here: [https://github.com/shuffle/shuffle-docs](https://github.com/shuffle/shuffle-docs). ### Setting up a local development environment Please follow the steps mentioned [here](https://github.com/Shuffle/Shuffle/blob/main/.github/install-guide.md#local-development-installation)! ## Related repositories -* OpenAPI apps: [https://github.com/frikky/security-openapis](https://github.com/frikky/security-openapis) -* Documentation: [https://github.com/frikky/shuffle-docs](https://github.com/frikky/shuffle-docs) -* Workflows: [https://github.com/frikky/shuffle-workflows](https://github.com/frikky/shuffle-workflows) -* Python apps: [https://github.com/frikky/shuffle-apps](https://github.com/frikky/shuffle-apps) +* OpenAPI apps: [https://github.com/shuffle/security-openapis](https://github.com/shuffle/security-openapis) +* Documentation: [https://github.com/shuffle/shuffle-docs](https://github.com/shuffle/shuffle-docs) +* Workflows: [https://github.com/shuffle/shuffle-workflows](https://github.com/shuffle/shuffle-workflows) +* Python apps: [https://github.com/shuffle/shuffle-apps](https://github.com/shuffle/python-apps) ## Features * Simple, feature rich [workflow editor](https://shuffler.io/docs/workflows) -* App creator using [OpenAPI](https://github.com/frikky/OpenAPI-security-definitions) +* App creator using [OpenAPI](https://github.com/shuffle/OpenAPI-security-definitions) * Premade apps for your security tools * Organization and sub-organization control * Hybrid resource sharing with shuffler.io (optional) @@ -66,7 +66,7 @@ Please follow the steps mentioned [here](https://github.com/Shuffle/Shuffle/blob [https://shuffler.io](https://shuffler.io) ## Contributing -We want to make the world of cybersecurity more accessible and need all the help we can get. Send an email to [frikky@shuffler](mailto:frikky@shuffler.io) and we'll make sure to give you any training you may need. +We want to make the world of cybersecurity more accessible and need all the help we can get. Send an email to [support@shuffler](mailto:support@shuffler.io) and we'll make sure to give you any training you may need. These are the main areas to contribute in: * Frontend (ReactJS) @@ -76,19 +76,19 @@ These are the main areas to contribute in: * Workflow creation (GUI & Conceptualizing) * Content Creation (Blogs, videos etc) -Contributing guidelines are outlined [here](https://github.com/frikky/Shuffle/blob/master/.github/CONTRIBUTING.md). +Contributing guidelines are outlined [here](https://github.com/shuffle/Shuffle/blob/master/.github/CONTRIBUTING.md). ## Contributors ![ICPL logo](https://github.com/Shuffle/Shuffle/blob/main/frontend/src/assets/img/icpl_logo.png) **Shuffle** - - + + -[**App magicians**](https://github.com/frikky/shuffle-apps) - - +[**App magicians**](https://github.com/shuffle/shuffle-apps) + + @@ -101,7 +101,7 @@ Shuffle backend: AGPLv3 Apps, specification and App SDK: MIT ## Architecture -![Shuffle Architecture](https://github.com/frikky/Shuffle/blob/main/frontend/src/assets/img/shuffle_architecture.png) +![Shuffle Architecture](https://github.com/shuffle/Shuffle/blob/main/frontend/src/assets/img/shuffle_architecture.png) ### Repository overview Below is the folder structure with a short explanation diff --git a/backend/go-app/go.mod b/backend/go-app/go.mod index 88015d5a..6e4b2ef2 100644 --- a/backend/go-app/go.mod +++ b/backend/go-app/go.mod @@ -2,7 +2,7 @@ module shuffle go 1.22.0 -//replace github.com/shuffle/shuffle-shared => ../../../shuffle-shared +// replace github.com/shuffle/shuffle-shared => ../../../shuffle-shared toolchain go1.22.2 @@ -24,6 +24,7 @@ require ( golang.org/x/crypto v0.22.0 google.golang.org/api v0.176.1 google.golang.org/grpc v1.63.2 + gopkg.in/src-d/go-git.v4 v4.13.1 gopkg.in/yaml.v3 v3.0.1 k8s.io/api v0.30.0 @@ -41,10 +42,12 @@ require ( github.com/Azure/go-ansiterm v0.0.0-20210617225240-d185dfc1b5a1 // indirect github.com/Masterminds/semver v1.5.0 // indirect github.com/Microsoft/go-winio v0.6.1 // indirect + github.com/ProtonMail/go-crypto v1.0.0 // indirect github.com/adrg/strutil v0.2.3 // indirect github.com/algolia/algoliasearch-client-go/v3 v3.18.1 // indirect github.com/bitly/go-simplejson v0.5.1 // indirect + github.com/bradfitz/gomemcache v0.0.0-20230905024940-24af94b03874 // indirect github.com/bradfitz/slice v0.0.0-20180809154707-2b758aa73013 // indirect github.com/cloudflare/circl v1.3.7 // indirect @@ -57,8 +60,10 @@ require ( github.com/docker/go-units v0.5.0 // indirect github.com/emicklei/go-restful/v3 v3.11.0 // indirect github.com/emirpasic/gods v1.18.1 // indirect + github.com/felixge/httpsnoop v1.0.4 // indirect github.com/frikky/schemaless v0.0.11 // indirect + github.com/go-git/gcfg v1.5.1-0.20230307220236-3a3c6141e376 // indirect github.com/go-logr/logr v1.4.1 // indirect github.com/go-logr/stdr v1.2.2 // indirect @@ -72,10 +77,12 @@ require ( github.com/google/go-github/v28 v28.1.1 // indirect github.com/google/go-querystring v1.0.0 // indirect github.com/google/gofuzz v1.2.0 // indirect + github.com/google/s2a-go v0.1.7 // indirect github.com/google/uuid v1.6.0 // indirect github.com/googleapis/enterprise-certificate-proxy v0.3.2 // indirect github.com/googleapis/gax-go/v2 v2.12.3 // indirect + github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 // indirect github.com/josharian/intern v1.0.0 // indirect github.com/json-iterator/go v1.1.12 // indirect @@ -99,9 +106,11 @@ require ( github.com/pjbgf/sha1cd v0.3.0 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/sashabaranov/go-openai v1.19.2 // indirect + github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3 // indirect github.com/sirupsen/logrus v1.9.3 // indirect github.com/skeema/knownhosts v1.2.2 // indirect + github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e // indirect github.com/src-d/gcfg v1.4.0 // indirect github.com/xanzy/ssh-agent v0.3.3 // indirect @@ -119,6 +128,7 @@ require ( golang.org/x/sys v0.19.0 // indirect golang.org/x/term v0.19.0 // indirect golang.org/x/text v0.14.0 // indirect + golang.org/x/time v0.5.0 // indirect golang.org/x/tools v0.18.0 // indirect google.golang.org/appengine v1.6.8 // indirect @@ -126,6 +136,7 @@ require ( google.golang.org/genproto/googleapis/api v0.0.0-20240314234333-6e1732d8331c // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20240415180920-8c6c420018be // indirect google.golang.org/protobuf v1.33.0 // indirect + gopkg.in/inf.v0 v0.9.1 // indirect gopkg.in/warnings.v0 v0.1.2 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect diff --git a/backend/go-app/main.go b/backend/go-app/main.go index 54f0874c..62870a45 100755 --- a/backend/go-app/main.go +++ b/backend/go-app/main.go @@ -5004,6 +5004,7 @@ func initHandlers() { r.HandleFunc("/api/v1/workflows/search", shuffle.HandleWorkflowRunSearch).Methods("POST", "OPTIONS") r.HandleFunc("/api/v1/workflows/schedules", shuffle.HandleGetSchedules).Methods("GET", "OPTIONS") r.HandleFunc("/api/v1/workflows/{key}/executions", shuffle.GetWorkflowExecutions).Methods("GET", "OPTIONS") + r.HandleFunc("/api/v1/workflows/{key}/executions/count", shuffle.HandleGetWorkflowRunCount).Methods("GET", "OPTIONS") r.HandleFunc("/api/v1/workflows/{key}/executions/{key}/rerun", checkUnfinishedExecution).Methods("GET", "POST", "OPTIONS") r.HandleFunc("/api/v1/workflows/{key}/executions/{key}/abort", shuffle.AbortExecution).Methods("GET", "OPTIONS") r.HandleFunc("/api/v1/workflows/{key}/schedule", scheduleWorkflow).Methods("POST", "OPTIONS") diff --git a/frontend/README.md b/frontend/README.md index fe883a45..571c5625 100755 --- a/frontend/README.md +++ b/frontend/README.md @@ -1,5 +1,6 @@ ## Localhost Certificate info: + Creating a localhost certificate: ``` diff --git a/frontend/run.sh b/frontend/run.sh index 9a608216..43ec3d89 100755 --- a/frontend/run.sh +++ b/frontend/run.sh @@ -1,18 +1,27 @@ #!/bin/sh + +TAG="nightly" + +if [ "$1" = "--latest" ]; then + echo "Using latest tag for image!" + TAG="latest" +fi + docker stop shuffle-frontend docker rm shuffle-frontend -#docker rmi ghcr.io/frikky/shuffle-frontend:nightly echo "Running build for website" -#sudo npm run build -docker build . -t ghcr.io/frikky/shuffle-frontend:nightly -docker tag ghcr.io/frikky/shuffle-frontend:nightly ghcr.io/shuffle/shuffle-frontend:nightly +# sudo npm run build +docker build . -t ghcr.io/frikky/shuffle-frontend:$TAG +docker tag ghcr.io/frikky/shuffle-frontend:$TAG ghcr.io/shuffle/shuffle-frontend:$TAG + +echo "Image built as ghcr.io/frikky/shuffle-frontend:$TAG!" echo "Starting server" # Rerun build locally for it to update :) docker run -it \ - -p 3001:80 \ - -p 3002:443 \ - -v $(pwd)/build:/usr/share/nginx/html:ro \ - --rm \ - ghcr.io/frikky/shuffle-frontend:nightly + -p 3001:80 \ + -p 3002:443 \ + -v $(pwd)/build:/usr/share/nginx/html:ro \ + --rm \ + ghcr.io/frikky/shuffle-frontend:$TAG diff --git a/frontend/src/components/AppFramework.jsx b/frontend/src/components/AppFramework.jsx index 0d31ce9d..00ef41c0 100644 --- a/frontend/src/components/AppFramework.jsx +++ b/frontend/src/components/AppFramework.jsx @@ -31,7 +31,9 @@ import { Delete as DeleteIcon, } from "@mui/icons-material"; + import edgehandles from "cytoscape-edgehandles"; + import cytoscape from "cytoscape"; import { toast } from 'react-toastify'; diff --git a/frontend/src/components/Oauth2Auth.jsx b/frontend/src/components/Oauth2Auth.jsx index c14e167d..40c043a4 100755 --- a/frontend/src/components/Oauth2Auth.jsx +++ b/frontend/src/components/Oauth2Auth.jsx @@ -362,6 +362,11 @@ const AuthenticationOauth2 = (props) => { } } + let workflowIdNew = "" + if (workflow !== undefined && workflow !== null && workflow.id !== undefined && workflow.id !== null) { + workflowIdNew = workflow.id + } + const appAuthData = { "label": "OAuth2 for " + selectedApp.name, "app": { diff --git a/frontend/src/views/AngularWorkflow.jsx b/frontend/src/views/AngularWorkflow.jsx index 69da2578..14f2b110 100755 --- a/frontend/src/views/AngularWorkflow.jsx +++ b/frontend/src/views/AngularWorkflow.jsx @@ -119,14 +119,18 @@ import { Polyline as PolylineIcon, QueryStats as QueryStatsIcon, AutoAwesome as AutoAwesomeIcon, + Add as AddIcon, ErrorOutline as ErrorOutlineIcon, + } from "@mui/icons-material"; //import * as cytoscape from "cytoscape"; import cytoscape from "cytoscape"; + import edgehandles from "cytoscape-edgehandles"; + import CytoscapeComponent from "react-cytoscapejs"; import Draggable from "react-draggable"; @@ -9325,6 +9329,7 @@ const AngularWorkflow = (defaultprops) => { newSelectedAction.example = newaction.returns.example; } + if ( newaction.description !== undefined && newaction.description !== null && @@ -15210,6 +15215,8 @@ const AngularWorkflow = (defaultprops) => { : null + + const RightsideBar = () => { const [hovered, setHovered] = useState(false) @@ -15270,6 +15277,59 @@ const AngularWorkflow = (defaultprops) => { return null } + useEffect(() => { + const handleKeyDown = (event) => { + if ((event.metaKey || event.ctrlKey) && event.key === '/') { + event.preventDefault(); // Prevent default browser behavior (like opening search bar) + if (!workflow.public && !executionRequestStarted) { + executeWorkflow(executionText, workflow.start, lastSaved); + } + } + if ((event.ctrlKey || event.metaKey) && event.key === "'") { + // Check if Ctrl (Windows/Linux) or Command (Mac) key is pressed along with '/' + if (!workflow.public && !executionModalOpen) { + setExecutionModalOpen(true); + getWorkflowExecution(props.match.params.key, ""); + } else if (!workflow.public && executionModalOpen) { + setExecutionModalOpen(false); + } + } + + if ((event.ctrlKey || event.metaKey) && event.key === "]") { + console.log("Show workflow revisions key pressed") + if (!workflow.public) { + setShowWorkflowRevisions(true) + setSelectedRevision(workflow) + //setOriginalWorkflow(workflow) + } + } + + if (( event.ctrlKey || event.metaKey ) && event.key === ";") { + if (!workflow.public && executionModalOpen) { + getWorkflowExecution(props.match.params.key, ""); + } + } + + if (( event.ctrlKey || event.metaKey ) && event.shiftKey) { + console.log("Shift key pressed") + if (!workflow.public && executionModalOpen) { + setExecutionRunning(false); + stop() + const cursearch = typeof window === "undefined" || window.location === undefined ? "" : window.location.search; + const newitem = removeParam("execution_id", cursearch); + navigate(curpath + newitem) + setExecutionModalView(0); + } + } + }; + + document.addEventListener('keydown', handleKeyDown); + + return () => { + document.removeEventListener('keydown', handleKeyDown); + }; + }, [executeWorkflow, executionText, workflow, lastSaved, executionRequestStarted]); + return (
{
) : (
+ { const { globalUrl, isLoaded, userdata, setUserData } = props; //const alert = useAlert(); @@ -61,6 +66,17 @@ const Settings = (props) => { }, 2000); } + const [showApiKey, setShowApiKey] = useState(false); + const [apiKeyCopied, setApiKeyCopied] = useState(false); + + const handleCopyApiKey = () => { + navigator.clipboard.writeText(userSettings.apikey); + setApiKeyCopied(true); + setTimeout(() => { + setApiKeyCopied(false); + }, 2000); + } + /* const [userdata.eth_info, setEthInfo] = useState(userdata.eth_info !== undefined && userdata.eth_info.account !== undefined && userdata.eth_info.account.length > 0 ? userdata.eth_info : { "account": "", @@ -965,6 +981,7 @@ const Settings = (props) => {

{passwordFormMessage}

+ {isCloud && ( <> @@ -1193,4 +1210,4 @@ const Settings = (props) => { return
{loadedCheck}
; }; -export default Settings; +export default Settings; \ No newline at end of file diff --git a/functions/onprem/orborus/orborus.go b/functions/onprem/orborus/orborus.go index cdface09..aef3e620 100755 --- a/functions/onprem/orborus/orborus.go +++ b/functions/onprem/orborus/orborus.go @@ -19,6 +19,7 @@ import ( "io" "io/ioutil" "log" + "math" "net" "net/http" "os" @@ -26,9 +27,8 @@ import ( "runtime" "strconv" "strings" - "time" "sync" - "math" + "time" //"os/signal" //"syscall" @@ -63,8 +63,8 @@ import ( var sleepTime = 2 // Making it work on low-end machines even during busy times :) -// May cause some things to run slowly -var maxConcurrency = 7 +// May cause some things to run slowly +var maxConcurrency = 7 // Timeout if something rashes var workerTimeoutEnv = os.Getenv("SHUFFLE_ORBORUS_EXECUTION_TIMEOUT") @@ -76,7 +76,8 @@ var dockerSwarmBridgeMTU = os.Getenv("SHUFFLE_SWARM_BRIDGE_DEFAULT_MTU") var dockerSwarmBridgeInterface = os.Getenv("SHUFFLE_SWARM_BRIDGE_DEFAULT_INTERFACE") var isKubernetes = os.Getenv("IS_KUBERNETES") var kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE") -var maxCPUPercent = 95 +var maxCPUPercent = 90 + // var baseimagename = "docker.pkg.github.com/shuffle/shuffle" // var baseimagename = "ghcr.io/frikky" @@ -482,6 +483,7 @@ func deployServiceWorkers(image string) { fmt.Sprintf("SHUFFLE_APP_SDK_TIMEOUT=%s", os.Getenv("SHUFFLE_APP_SDK_TIMEOUT")), fmt.Sprintf("SHUFFLE_MAX_SWARM_NODES=%d", os.Getenv("SHUFFLE_MAX_SWARM_NODES")), fmt.Sprintf("SHUFFLE_BASE_IMAGE_NAME=%s", os.Getenv("SHUFFLE_BASE_IMAGE_NAME")), + fmt.Sprintf("SHUFFLE_APP_REQUEST_TIMEOUT=%s", os.Getenv("SHUFFLE_APP_REQUEST_TIMEOUT")), }, //Hosts: []string{ // innerContainerName, @@ -781,9 +783,9 @@ func deployWorker(image string, identifier string, env []string, executionReques log.Printf("[INFO] Created pod %q in namespace %q\n", createdPod.Name, createdPod.Namespace) return nil - } + } - // Binds is the actual "-v" volume. + // Binds is the actual "-v" volume. // Max 20% CPU every second //CPUQuota: 25000, @@ -1138,7 +1140,7 @@ func getOrborusStats(ctx context.Context) shuffle.OrborusStats { newStats.MaxQueue = maxConcurrency newStats.Queue = executionCount - if isKubernetes == "true" || runningMode == "kubernetes" || runningMode == "k8s" { + if isKubernetes == "true" || runningMode == "kubernetes" || runningMode == "k8s" { newStats.Kubernetes = true return newStats } @@ -1171,6 +1173,7 @@ func getOrborusStats(ctx context.Context) shuffle.OrborusStats { // Get list of all running containers containers, err := dockercli.ContainerList(ctx, container.ListOptions{}) + if err != nil { log.Printf("[ERROR] Failed getting container list: %s", err) return newStats @@ -1229,33 +1232,33 @@ func getOrborusStats(ctx context.Context) shuffle.OrborusStats { // check if it's NaN or Inf if !math.IsNaN(result.cpuUsage) { totalCPU += float64(result.cpuUsage) - } + } if !math.IsNaN(result.memoryUsage) { memUsage += float64(result.memoryUsage) } } - newStats.CPUPercent = totalCPU/float64(newStats.CPU) + newStats.CPUPercent = totalCPU / float64(newStats.CPU) newStats.MemoryPercent = memUsage //log.Printf("[DEBUG] CPU: %.2f, Memory: %.2f", newStats.CPUPercent, newStats.MemoryPercent) /* - cpuPercent, err := cpu.Percent(250*time.Millisecond, false) - if err == nil && len(cpuPercent) > 0 { - newStats.CPUPercent = cpuPercent[0] - } - //Percent(interval time.Duration, percpu bool) ([]float64, error) + cpuPercent, err := cpu.Percent(250*time.Millisecond, false) + if err == nil && len(cpuPercent) > 0 { + newStats.CPUPercent = cpuPercent[0] + } + //Percent(interval time.Duration, percpu bool) ([]float64, error) - // Get memory usage - memory, err := memory.Get() - if err != nil { - log.Printf("[ERROR] Failed getting memory stats: %s", err) - } else { - newStats.Memory = int(memory.Used) - newStats.MaxMemory = int(memory.Total) - } + // Get memory usage + memory, err := memory.Get() + if err != nil { + log.Printf("[ERROR] Failed getting memory stats: %s", err) + } else { + newStats.Memory = int(memory.Used) + newStats.MaxMemory = int(memory.Total) + } */ // Get disk usage @@ -1560,13 +1563,20 @@ func main() { } if os.Getenv("SHUFFLE_MAX_CPU") != "" { - // parse + // parse tmpInt, err := strconv.Atoi(os.Getenv("SHUFFLE_MAX_CPU")) if err == nil { maxCPUPercent = tmpInt } } + swarmPollingTime := time.Now() + swarmRequestsMade := 0 + swarmControlMode := false + if os.Getenv("SHUFFLE_SWARM_CONTROL_MODE") == "true" { + swarmControlMode = true + } + log.Printf("[INFO] Waiting for executions at %s with Environment %#v", fullUrl, environment) hasStarted := false for { @@ -1732,6 +1742,20 @@ 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")) { + if len(executionRequests.Data) > 50 { + executionRequests.Data = executionRequests.Data[0:50] + } + + if swarmRequestsMade > 100 && time.Since(swarmPollingTime).Seconds() > 5 { + log.Printf("[DEBUG] Swarm requests made: %d", swarmRequestsMade) + time.Sleep(time.Duration(sleepTime) * time.Second) + + swarmPollingTime = time.Now() + swarmRequestsMade = 0 + } + + swarmRequestsMade += len(executionRequests.Data) } // New, abortable version. Should check executionid and remove everything else @@ -1760,9 +1784,9 @@ func main() { // Should check when last this was ran, and if it's more than 10 minutes ago and it's not finished, we should run it again? /* - if swarmConfig != "run" && swarmConfig != "swarm" { - continue - } + if swarmConfig != "run" && swarmConfig != "swarm" { + continue + } */ } @@ -1818,7 +1842,6 @@ func main() { env = append(env, fmt.Sprintf("SHUFFLE_VOLUME_BINDS=%s", os.Getenv("SHUFFLE_VOLUME_BINDS"))) } - if len(os.Getenv("SHUFFLE_APP_SDK_TIMEOUT")) > 0 { env = append(env, fmt.Sprintf("SHUFFLE_APP_SDK_TIMEOUT=%s", os.Getenv("SHUFFLE_APP_SDK_TIMEOUT"))) } @@ -2585,6 +2608,7 @@ func getRunningWorkers(ctx context.Context, workerTimeout int) int { if isKubernetes == "true" { log.Printf("[INFO] Getting running workers in kubernetes") + thresholdTime := time.Now().Add(time.Duration(-workerTimeout) * time.Second) clientset, _, err := getKubernetesClient() @@ -2616,20 +2640,20 @@ func getRunningWorkers(ctx context.Context, workerTimeout int) int { containers, err := dockercli.ContainerList(ctx, container.ListOptions{ All: true, }) - + // Automatically updates the version if err != nil { log.Printf("[ERROR] Error getting containers: %s", err) - + newVersionSplit := strings.Split(fmt.Sprintf("%s", err), "version is") if len(newVersionSplit) > 1 { //dockerApiVersion = strings.TrimSpace(newVersionSplit[1]) log.Printf("[DEBUG] WANT to change the API version to default to %s?", strings.TrimSpace(newVersionSplit[1])) } - + return maxConcurrency } - + currenttime := time.Now().Unix() for _, container := range containers { @@ -2642,7 +2666,7 @@ func getRunningWorkers(ctx context.Context, workerTimeout int) int { break } } - + // Check image name if !shuffleFound { continue @@ -2650,13 +2674,13 @@ func getRunningWorkers(ctx context.Context, workerTimeout int) int { //} else { // log.Printf("NAME: %s", container.Image) } - + for _, name := range container.Names { // FIXME - add name_version_uid_uid regex check as well if !strings.HasPrefix(name, "/worker") { continue } - + //log.Printf("Time: %d - %d", currenttime-container.Created, int64(workerTimeout)) if container.State == "running" && currenttime-container.Created < int64(workerTimeout) { counter += 1 @@ -2800,7 +2824,7 @@ func sendWorkerRequest(workflowExecution shuffle.ExecutionRequest) error { streamUrl = fmt.Sprintf("%s:33333/api/v1/execute", parsedBaseurl) } - if len(workerServerUrl) > 0 { + if len(workerServerUrl) > 0 { streamUrl = fmt.Sprintf("%s:33333/api/v1/execute", workerServerUrl) } diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 3ac2ca93..89153b11 100644 --- a/functions/onprem/worker/worker.go +++ b/functions/onprem/worker/worker.go @@ -3,7 +3,6 @@ package main import ( "github.com/shuffle/shuffle-shared" - "bytes" "context" "encoding/json" @@ -22,8 +21,8 @@ import ( "time" "github.com/docker/docker/api/types" - "github.com/docker/docker/api/types/filters" "github.com/docker/docker/api/types/container" + "github.com/docker/docker/api/types/filters" "github.com/docker/docker/api/types/mount" dockerclient "github.com/docker/docker/client" // This is for automatic removal of certain code :) @@ -57,6 +56,7 @@ var kubernetesNamespace = os.Getenv("KUBERNETES_NAMESPACE") // var baseimagename = os.Getenv("SHUFFLE_BASE_IMAGE_NAME") + // var baseimagename = "registry.hub.docker.com/frikky/shuffle" var registryName = "registry.hub.docker.com" var sleepTime = 2 @@ -81,6 +81,7 @@ var startAction string //var allLogs map[string]string //var containerIds []string var downloadedImages []string + type ImageDownloadBody struct { Image string `json:"image"` } @@ -92,7 +93,6 @@ type ImageRequest struct { var finishedExecutions []string var imagesDistributed []string - // Images to be autodeployed in the latest version of Shuffle. var autoDeploy = map[string]string{ "http:1.4.0": "frikky/shuffle:http_1.4.0", @@ -139,7 +139,6 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo return err } - handleExecutionResult(workflowExecution) validated := shuffle.ValidateFinished(ctx, -1, workflowExecution) if validated { @@ -179,7 +178,7 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo } } - if len(subflowId) == 0 { + if len(subflowId) == 0 { log.Printf("[DEBUG][%s] No waiting result found. Not polling", workflowExecution.ExecutionId) for _, action := range workflowExecution.Workflow.Actions { @@ -187,19 +186,17 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo workflowExecution.Workflow.Triggers = append(workflowExecution.Workflow.Triggers, shuffle.Trigger{ AppName: action.AppName, Parameters: action.Parameters, - ID: action.ID, + ID: action.ID, }) } } - for _, trigger := range workflowExecution.Workflow.Triggers { //log.Printf("[DEBUG] Found trigger %s", trigger.AppName) if trigger.AppName != "User Input" && trigger.AppName != "Shuffle Workflow" && trigger.AppName != "shuffle-subflow" { continue } - // check if it has wait for results in params wait := false for _, param := range trigger.Parameters { @@ -218,9 +215,9 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo //log.Printf("[DEBUG][%s] Found result %s", workflowExecution.ExecutionId, result.Action.ID) if result.Action.ID == trigger.ID && result.Status != "SUCCESS" && result.Status != "FAILURE" { //log.Printf("[DEBUG][%s] Found subflow result that is not handled. Waiting for results", workflowExecution.ExecutionId) - + subflowId = result.Action.ID - found = true + found = true break } } @@ -239,21 +236,20 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo if len(subflowId) > 0 { // Under rerun period timeout - timeComparison := 120 + timeComparison := 120 log.Printf("[DEBUG][%s] Starting polling for %d seconds to see if new subflow updates are found on the backend that are not handled. Subflow ID: %s", workflowExecution.ExecutionId, timeComparison, subflowId) timestart := time.Now() streamResultUrl := fmt.Sprintf("%s/api/v1/streams/results", baseUrl) for { - err = handleSubflowPoller(ctx, workflowExecution, streamResultUrl, subflowId) + err = handleSubflowPoller(ctx, workflowExecution, streamResultUrl, subflowId) if err == nil { log.Printf("[DEBUG] Subflow is finished and we are breaking the thingy") - + if os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" && workflowExecution.ExecutionSource != "default" { log.Printf("[DEBUG] Force shutdown of worker due to optimized run with webserver. Expecting reruns to take care of this") os.Exit(0) } - break } @@ -275,7 +271,6 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo return nil } - // removes every container except itself (worker) func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason string, handleResultSend bool) { log.Printf("[DEBUG][%s] Shutdown (%s) started with reason %#v. Result amount: %d. ResultsSent: %d, Send result: %#v, Parent: %#v", workflowExecution.ExecutionId, workflowExecution.Status, reason, len(workflowExecution.Results), requestsSent, handleResultSend, workflowExecution.ExecutionParent) @@ -315,7 +310,7 @@ func shutdown(workflowExecution shuffle.WorkflowExecution, nodeId string, reason } */ } else { - + } if len(reason) > 0 && len(nodeId) > 0 { @@ -504,7 +499,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] if !strings.Contains(param.Value, "shuffle-backend") { continue - } + } // Automatic replacement as this is default if len(os.Getenv("BASE_URL")) > 0 { @@ -519,7 +514,6 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] } } - // Max 10% CPU every second //CPUShares: 128, //CPUQuota: 10000, @@ -546,7 +540,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] // Get environment for certificates volumeBinds := []string{} - volumeBindString:= os.Getenv("SHUFFLE_VOLUME_BINDS") + volumeBindString := os.Getenv("SHUFFLE_VOLUME_BINDS") if len(volumeBindString) > 0 { volumeBindSplit := strings.Split(volumeBindString, ",") for _, volumeBind := range volumeBindSplit { @@ -587,7 +581,6 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env [] Env: env, } - // Checking as late as possible, just in case. newExecId := fmt.Sprintf("%s_%s", workflowExecution.ExecutionId, action.ID) _, err := shuffle.GetCache(ctx, newExecId) @@ -867,7 +860,7 @@ func askOtherWorkersToDownloadImage(image string) { // Check environment SHUFFLE_AUTO_IMAGE_DOWNLOAD if os.Getenv("SHUFFLE_AUTO_IMAGE_DOWNLOAD") == "false" { log.Printf("[DEBUG] SHUFFLE_AUTO_IMAGE_DOWNLOAD is false. NOT distributing images %s", image) - return + return } if shuffle.ArrayContains(imagesDistributed, image) { @@ -900,7 +893,7 @@ func askOtherWorkersToDownloadImage(image string) { req, err := http.NewRequest( "POST", url, - bytes.NewBuffer(imageJSON), + bytes.NewBuffer(imageJSON), ) if err != nil { @@ -940,7 +933,6 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { return } - startAction, extra, children, parents, visited, executed, nextActions, environments := shuffle.GetExecutionVariables(ctx, workflowExecution.ExecutionId) dockercli, err := dockerclient.NewEnvClient() @@ -1008,7 +1000,7 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { // marshal action and put it in there rofl //log.Printf("[INFO][%s] Time to execute %s (%s) with app %s:%s, function %s, env %s with %d parameters.", workflowExecution.ExecutionId, action.ID, action.Label, action.AppName, action.AppVersion, action.Name, action.Environment, len(action.Parameters)) - + log.Printf("[DEBUG][%s] Action: Send, Label: '%s', Action: '%s', Run status: %s, Extra=", workflowExecution.ExecutionId, action.Label, action.AppName, workflowExecution.Status) actionData, err := json.Marshal(action) @@ -1094,10 +1086,9 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { } if len(os.Getenv("SHUFFLE_APP_SDK_TIMEOUT")) > 0 { - env = append(env, fmt.Sprintf("SHUFFLE_APP_SDK_TIMEOUT=%s", os.Getenv("SHUFFLE_APP_SDK_TIMEOUT"))) + env = append(env, fmt.Sprintf("SHUFFLE_APP_SDK_TIMEOUT=%s", os.Getenv("SHUFFLE_APP_SDK_TIMEOUT"))) } - // Fixes issue: // standard_go init_linux.go:185: exec user process caused "argument list too long" // https://devblogs.microsoft.com/oldnewthing/20100203-00/?p=15083 @@ -1125,8 +1116,6 @@ func handleExecutionResult(workflowExecution shuffle.WorkflowExecution) { fmt.Sprintf("%s:%s_%s", baseimagename, parsedAppname, action.AppVersion), } - - // If cleanup is set, it should run for efficiency pullOptions := types.ImagePullOptions{} if cleanupEnv == "true" { @@ -1629,7 +1618,6 @@ func handleSubflowPoller(ctx context.Context, workflowExecution shuffle.Workflow } } - if workflowExecution.Status == "WAITING" && workflowExecution.ExecutionSource != "default" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm" { log.Printf("[INFO][%s] Workflow execution is waiting. Exiting worker, as backend will restart it.", workflowExecution.ExecutionId) shutdown(workflowExecution, "", "", true) @@ -2062,10 +2050,9 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl resp.Write([]byte(fmt.Sprintf(`{"success": true, "reason": "Execution is not executing, but %s"}`, workflowExecution.Status))) } - log.Printf("[DEBUG][%s] Shutting down (35)", workflowExecution.ExecutionId) - // Force sending result + // Force sending result shutdownData, err := json.Marshal(workflowExecution) if err != nil { log.Printf("[ERROR][%s] Failed marshalling execution (35): %s", workflowExecution.ExecutionId, err) @@ -2141,11 +2128,11 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl attempts += 1 log.Printf("[DEBUG][%s] Rerunning transaction as results has changed. %d vs %d", workflowExecution.ExecutionId, len(parsedValue.Results), resultLength) /* - if len(workflowExecution.Results) <= len(workflowExecution.Workflow.Actions) { - log.Printf("[DEBUG][%s] Rerunning transaction as results has changed. %d vs %d", workflowExecution.ExecutionId, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)) - runWorkflowExecutionTransaction(ctx, attempts, workflowExecutionId, actionResult, resp) - return - } + if len(workflowExecution.Results) <= len(workflowExecution.Workflow.Actions) { + log.Printf("[DEBUG][%s] Rerunning transaction as results has changed. %d vs %d", workflowExecution.ExecutionId, len(workflowExecution.Results), len(workflowExecution.Workflow.Actions)) + runWorkflowExecutionTransaction(ctx, attempts, workflowExecutionId, actionResult, resp) + return + } */ } } @@ -2178,8 +2165,6 @@ func runWorkflowExecutionTransaction(ctx context.Context, attempts int64, workfl } func sendSelfRequest(actionResult shuffle.ActionResult) { - - data, err := json.Marshal(actionResult) if err != nil { log.Printf("[ERROR][%s] Shutting down (24): Failed to unmarshal data for backend: %s", actionResult.ExecutionId, err) @@ -2238,10 +2223,10 @@ func sendResult(workflowExecution shuffle.WorkflowExecution, data []byte) { // Basically to reduce backend strain /* - if shuffle.ArrayContains(finishedExecutions, workflowExecution.ExecutionId) { - log.Printf("[INFO][%s] NOT sending backend info since it's already been sent before.", workflowExecution.ExecutionId) - return - } + if shuffle.ArrayContains(finishedExecutions, workflowExecution.ExecutionId) { + log.Printf("[INFO][%s] NOT sending backend info since it's already been sent before.", workflowExecution.ExecutionId) + return + } */ // Take it down again @@ -2252,6 +2237,7 @@ func sendResult(workflowExecution shuffle.WorkflowExecution, data []byte) { } finishedExecutions = append(finishedExecutions, workflowExecution.ExecutionId) + */ streamUrl := fmt.Sprintf("%s/api/v1/streams", baseUrl) @@ -2308,6 +2294,7 @@ func sendResult(workflowExecution shuffle.WorkflowExecution, data []byte) { if workflowExecution.Status == "FINISHED" || workflowExecution.Status == "ABORTED" || (len(environments) == 1 && requestsSent == 0 && len(workflowExecution.Results) >= 1 && os.Getenv("SHUFFLE_SWARM_CONFIG") != "run" && os.Getenv("SHUFFLE_SWARM_CONFIG") != "swarm") || (len(workflowExecution.Results) >= len(workflowExecution.Workflow.Actions)+extra && len(workflowExecution.Workflow.Actions) > 0) { + if workflowExecution.Status == "FINISHED" { for _, result := range workflowExecution.Results { if result.Status == "EXECUTING" || result.Status == "WAITING" { @@ -2316,8 +2303,7 @@ func sendResult(workflowExecution shuffle.WorkflowExecution, data []byte) { } } } - - + log.Printf("[DEBUG][%s] Should send full result to %s", workflowExecution.ExecutionId, baseUrl) //data = fmt.Sprintf(`{"execution_id": "%s", "authorization": "%s"}`, executionId, authorization) @@ -2401,7 +2387,7 @@ func handleGetStreamResults(resp http.ResponseWriter, request *http.Request) { // GetLocalIP returns the non loopback local IP of the host func getLocalIP() string { - + addrs, err := net.InterfaceAddrs() if err != nil { return "" @@ -2446,8 +2432,7 @@ func webserverSetup(workflowExecution shuffle.WorkflowExecution) net.Listener { } log.Printf("[DEBUG] OLD HOSTNAME: %s", appCallbackUrl) - - + port := listener.Addr().(*net.TCPAddr).Port // Set the port environment variable os.Setenv("WORKER_PORT", fmt.Sprintf("%d", port)) @@ -2605,7 +2590,6 @@ func downloadDockerImageBackend(client *http.Client, imageName string) error { */ } - // Runs data discovery func sendAppRequest(ctx context.Context, incomingUrl, appName string, port int, action *shuffle.Action, workflowExecution *shuffle.WorkflowExecution) error { @@ -2869,7 +2853,7 @@ func getStreamResultsWrapper(client *http.Client, req *http.Request, workflowExe if newresp.StatusCode != 200 { log.Printf("[ERROR] %sStatusCode (1): %d", string(body), newresp.StatusCode) time.Sleep(time.Duration(sleepTime) * time.Second) - return environments, errors.New(fmt.Sprintf("Bad status code: %d", newresp.StatusCode) ) + return environments, errors.New(fmt.Sprintf("Bad status code: %d", newresp.StatusCode)) } err = json.Unmarshal(body, &workflowExecution) @@ -2952,7 +2936,6 @@ func getStreamResultsWrapper(client *http.Client, req *http.Request, workflowExe // Set environment variable - //log.Printf("Before wait") //wg := sync.WaitGroup{} //wg.Add(1) @@ -3024,7 +3007,6 @@ func main() { swarmConfig := os.Getenv("SHUFFLE_SWARM_CONFIG") log.Printf("[INFO] Running with timezone %s and swarm config %#v", timezone, swarmConfig) - authorization := "" executionId := "" @@ -3329,7 +3311,6 @@ func handleDownloadImage(resp http.ResponseWriter, request *http.Request) { return } - for _, img := range images { for _, tag := range img.RepoTags { splitTag := strings.Split(tag, ":") @@ -3342,7 +3323,7 @@ func handleDownloadImage(resp http.ResponseWriter, request *http.Request) { possibleNames = append(possibleNames, fmt.Sprintf("frikky/shuffle:%s", baseTag)) possibleNames = append(possibleNames, fmt.Sprintf("registry.hub.docker.com/frikky/shuffle:%s", baseTag)) - if (arrayContains(possibleNames, image.Image)) { + if arrayContains(possibleNames, image.Image) { log.Printf("[DEBUG] Image %s already downloaded that has been requested to download", image.Image) resp.WriteHeader(200) resp.Write([]byte(fmt.Sprintf(`{"success": false, "reason": "image already present"}`))) @@ -3367,7 +3348,6 @@ func runWebserver(listener net.Listener) { r.HandleFunc("/api/v1/run", handleRunExecution).Methods("POST", "OPTIONS") r.HandleFunc("/api/v1/download", handleDownloadImage).Methods("POST", "OPTIONS") - if strings.ToLower(os.Getenv("SHUFFLE_DEBUG_MEMORY")) == "true" { r.HandleFunc("/debug/pprof/", pprof.Index) r.HandleFunc("/debug/pprof/heap", pprof.Handler("heap").ServeHTTP)