diff --git a/functions/onprem/worker/worker.go b/functions/onprem/worker/worker.go index 62590dfb..506d7854 100644 --- a/functions/onprem/worker/worker.go +++ b/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) } -} +} \ No newline at end of file