feat: publically releasing scale worker (updated)

functions/onprem/worker/worker.go

@@ -32,9 +32,9 @@ import (
 	uuid "github.com/satori/go.uuid"
 
 	//k8s deps
-	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	appsv1 "k8s.io/api/apps/v1"
 	"k8s.io/apimachinery/pkg/util/intstr"
 	"k8s.io/client-go/kubernetes"
 )
@@ -138,6 +138,7 @@ func setWorkflowExecution(ctx context.Context, workflowExecution shuffle.Workflo
 		return err
 	}
 
+
 	handleExecutionResult(workflowExecution)
 	validated := shuffle.ValidateFinished(ctx, -1, workflowExecution)
 	if validated {
@@ -515,7 +516,7 @@ func deployk8sApp(image string, identifier string, env []string) error {
 	// }
 
 	// use deployment instead of pod
-	// then expose a service similarly.
+	// then expose a service similarly. 
 	// number of replicas can be set to os.Getenv("SHUFFLE_SCALE_REPLICAS")
 	replicaNumberStr := os.Getenv("SHUFFLE_SCALE_REPLICAS")
 	replicaNumber := 1
@@ -525,7 +526,7 @@ func deployk8sApp(image string, identifier string, env []string) error {
 			log.Printf("[ERROR] %s is not a valid number for replication", replicaNumberStr)
 		} else {
 			replicaNumber = tmpInt
-
+		
 		}
 	}
 
@@ -737,6 +738,7 @@ func deployApp(cli *dockerclient.Client, image string, identifier string, env []
 		}
 	}
 
+
 	// Max 10% CPU every second
 	//CPUShares: 128,
 	//CPUQuota:  10000,
@@ -2389,6 +2391,7 @@ 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)
@@ -2526,6 +2529,7 @@ func validateFinished(workflowExecution shuffle.WorkflowExecution) bool {
 			}
 		}
 
+
 		log.Printf("[DEBUG][%s] Should send full result to %s", workflowExecution.ExecutionId, baseUrl)
 
 		//data = fmt.Sprintf(`{"execution_id": "%s", "authorization": "%s"}`, executionId, authorization)
@@ -2609,6 +2613,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 ""
@@ -2658,6 +2663,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))
@@ -2815,6 +2821,7 @@ func findActiveSwarmNodes(dockercli *dockerclient.Client) (int64, error) {
 	*/
 }
 
+
 // Runs data discovery
 
 func sendAppRequest(ctx context.Context, incomingUrl, appName string, port int, action *shuffle.Action, workflowExecution *shuffle.WorkflowExecution) error {
@@ -3244,6 +3251,7 @@ func main() {
 	swarmConfig := os.Getenv("SHUFFLE_SWARM_CONFIG")
 	log.Printf("[INFO] Running with timezone %s and swarm config %#v", timezone, swarmConfig)
 
+
 	authorization := ""
 	executionId := ""
 
@@ -3585,6 +3593,7 @@ 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)
@@ -3602,6 +3611,7 @@ func runWebserver(listener net.Listener) {
 
 	//log.Fatal(http.Serve(listener, nil))
 
+
 	log.Printf("[DEBUG] NEW webserver setup")
 
 	http.Handle("/", r)
@@ -3617,4 +3627,4 @@ func runWebserver(listener net.Listener) {
 	if err != nil {
 		log.Printf("[ERROR] Serve issue in worker: %#v", err)
 	}
-}
+}