Merge pull request #37 from getsumio/develop

Develop

Author: getsumio

VERSION

@@ -1 +1 @@
-v2.0
+v2.0.1

go.mod

@@ -3,6 +3,7 @@ module github.com/getsumio/getsum
 go 1.13
 
 require (
+	github.com/google/uuid v1.1.1
 	github.com/logrusorgru/aurora v0.0.0-20191017060258-dc85c304c434
 	github.com/mitchellh/go-homedir v1.1.0
 	golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550

go.sum

@@ -1,4 +1,6 @@
 github.com/go-logr/logr v0.1.0/go.mod h1:ixOQHD9gLJUVQQ2ZOR7zLEifBX6tGkNJF4QyIY7sIas=
+github.com/google/uuid v1.1.1 h1:Gkbcsh/GbpXz7lPftLA3P6TYMwjCLYm83jiFQZF/3gY=
+github.com/google/uuid v1.1.1/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/kisielk/errcheck v1.2.0 h1:reN85Pxc5larApoH1keMBiu2GWtPqXQ1nc9gx+jOU+E=
 github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
 github.com/logrusorgru/aurora v0.0.0-20191017060258-dc85c304c434 h1:im9kkmH0WWwxzegiv18gSUJbuXR9y028rXrWuPp6Jug=

internal/config/config.go

@@ -2,28 +2,26 @@ package config
 
 //config dto
 type Config struct {
-	File         *string       `json:"file"`
-	LocalOnly    *bool         `json:"local_only"`
-	Proxy        *string       `json:"proxy"`
-	Algorithm    []string      `json:"algorithm"`
-	Cheksum      *string       `json:"cheksum"`
-	RemoteOnly   *bool         `json:"remote_only"`
-	OnlyChecksum *bool         `json:"only_checksum"`
-	LogLevel     *string       `json:"log_level"`
-	Timeout      *int          `json:"timeout"`
-	All          *bool         `json:"all"`
-	Key          *string       `json:"key"`
-	Supplier     *string       `json:"supplier"`
-	OnFailure    *string       `json:"on_failure"`
-	Serve        *bool         `json:"serve"`
-	Listen       *string       `json:"listen"`
-	Port         *int          `json:"port"`
-	Servers      ServerConfigs `json:"servers"`
-	Dir          *string       `json:"dir"`
-	TLSKey       *string       `json:"tls_key"`
-	TLSCert      *string       `json:"tls_cert"`
-	ServerConfig *string       `json:"server_config"`
-	Keep         *bool         `json:"keep"`
+	File         *string `json:"file"`
+	LocalOnly    *bool
+	Proxy        *string  `json:"proxy"`
+	Algorithm    []string `json:"algorithm"`
+	Cheksum      *string  `json:"cheksum"`
+	RemoteOnly   *bool
+	LogLevel     *string
+	Timeout      *int    `json:"timeout"`
+	All          *bool   `json:"all"`
+	Key          *string `json:"key"`
+	Supplier     *string `json:"supplier"`
+	Serve        *bool
+	Listen       *string
+	Port         *int
+	Servers      ServerConfigs
+	Dir          *string
+	TLSKey       *string
+	TLSCert      *string
+	ServerConfig *string
+	Keep         *bool
 }
 
 //this is for collecting server info from yaml files

internal/provider/onpremiseprovider.go

@@ -3,6 +3,7 @@ package providers
 import (
 	"bytes"
 	"encoding/json"
+	"fmt"
 	"net/http"
 
 	"github.com/getsumio/getsum/internal/config"
@@ -22,6 +23,7 @@ type RemoteProvider struct {
 	address     string
 	status      *status.Status
 	hasRunError bool
+	processId   string
 }
 
 //notifies server to run
@@ -52,6 +54,7 @@ func remoteRun(l *RemoteProvider) {
 	if err != nil {
 		//set error as provider status
 		//status() method will handle
+		logger.Debug("Run response has an error: %s", err.Error())
 		setErrorStatus(err, l)
 		l.hasRunError = true
 		return
@@ -61,10 +64,15 @@ func remoteRun(l *RemoteProvider) {
 	decoder := json.NewDecoder(resp.Body)
 	err = decoder.Decode(l.status)
 	if err != nil {
+		logger.Debug("Run response received for %s but got an error: %s", l.Name, err.Error())
 		setErrorStatus(err, l)
 		l.hasRunError = true
 	}
 	l.hasRunError = l.status.Type == status.ERROR
+	if !l.hasRunError {
+		logger.Trace("Process id for runner: %s", l.status.Value)
+		l.processId = l.status.Value
+	}
 
 }
 
@@ -77,8 +85,9 @@ func closeResponse(response *http.Response) {
 
 //fetches server using GET and collects its status
 func remoteStatus(l *RemoteProvider) *status.Status {
+	statusAddress := fmt.Sprintf("%s/%s", l.address, l.processId)
 	//reach the server
-	resp, err := l.client.Get(l.address)
+	resp, err := l.client.Get(statusAddress)
 	if err != nil {
 		setErrorStatus(err, l)
 		return l.status
@@ -99,7 +108,8 @@ func remoteStatus(l *RemoteProvider) *status.Status {
 //trigger termination on remote server using http DELETE
 func remoteTerminate(l *RemoteProvider) error {
 	//let the server know process terminated
-	req, err := http.NewRequest("DELETE", l.address, nil)
+	deleteAddress := fmt.Sprintf("%s/%s", l.address, l.processId)
+	req, err := http.NewRequest("DELETE", deleteAddress, nil)
 	if err != nil {
 		return err
 	}

internal/servers/onpremise.go

@@ -3,7 +3,10 @@ package servers
 import (
 	"encoding/json"
 	"fmt"
+	"html"
 	"net/http"
+	"path"
+	"regexp"
 	"sync"
 
 	"github.com/getsumio/getsum/internal/config"
@@ -12,22 +15,32 @@ import (
 	"github.com/getsumio/getsum/internal/status"
 	. "github.com/getsumio/getsum/internal/supplier"
 	"github.com/getsumio/getsum/internal/validation"
+	"github.com/google/uuid"
 )
 
 //server instance to run in server mode
 type OnPremiseServer struct {
 	StoragePath string
-	Supplier    Supplier
-	mux         sync.Mutex
+	mux         *sync.RWMutex
+	suppliers   map[string]Supplier
 }
 
 var factory ISupplierFactory
 
+const uuidPattern = "^[a-fA-F0-9]{8}-[a-fA-F0-9]{4}-4[a-fA-F0-9]{3}-[8|9|aA|bB][a-fA-F0-9]{3}-[a-fA-F0-9]{12}$"
+
+var regex *regexp.Regexp = regexp.MustCompile(uuidPattern)
+
+const default_capacity = 250
+const threshold = 150
+
 //start server in given config listen address and port or tls details
 //TODO add interface support
 func (s *OnPremiseServer) Start(config *config.Config) error {
 	logger.Level = logger.LevelInfo
 	factory = new(SupplierFactory)
+	s.suppliers = make(map[string]Supplier)
+	s.mux = &sync.RWMutex{}
 	http.HandleFunc("/", s.handle)
 	listenAddress := fmt.Sprintf("%s:%d", *config.Listen, *config.Port)
 	var err error
@@ -45,14 +58,18 @@ func (s *OnPremiseServer) Start(config *config.Config) error {
 
 //get executed to reach status
 //so collect status if there is any runner
-func handleGet(s *OnPremiseServer, w http.ResponseWriter, r *http.Request) {
+func handleGet(s *OnPremiseServer, w http.ResponseWriter, r *http.Request, id string) {
 	//check if any runner
-	if s.Supplier == nil {
+	s.mux.RLock()
+	defer s.mux.RUnlock()
+	logger.Info("Checking if there is a process with id : %s", id)
+	supplier, ok := s.suppliers[id]
+	if !ok {
 		handleError("There is no running process", w)
 		return
 	}
 	//collect status and return
-	stat := s.Supplier.Status()
+	stat := supplier.Status()
 	status, err := json.Marshal(stat)
 	if err != nil {
 		handleError("System can not parse given status %s", w, err.Error())
@@ -64,14 +81,8 @@ func handleGet(s *OnPremiseServer, w http.ResponseWriter, r *http.Request) {
 
 //post executed to Run a new calculation
 func handlePost(s *OnPremiseServer, w http.ResponseWriter, r *http.Request) {
-	//check if any runner
-	if s.Supplier != nil {
-		stat := s.Supplier.Status()
-		if stat.Type <= status.RUNNING {
-			handleError("Server already running another process", w)
-			return
-		}
-	}
+	s.mux.Lock()
+	defer s.mux.Unlock()
 	//read the config from request
 	jsonDecoder := json.NewDecoder(r.Body)
 	config := &Config{}
@@ -85,49 +96,73 @@ func handlePost(s *OnPremiseServer, w http.ResponseWriter, r *http.Request) {
 	if err != nil {
 		handleError(err.Error(), w)
 		return
-
 	}
 
 	//get supplier instance, only single algo supported on server mode
 	var algorithm = ValueOf(&config.Algorithm[0])
 	config.Dir = &s.StoragePath
-	s.Supplier, err = factory.GetSupplierByAlgo(config, &algorithm)
+	supplier, err := factory.GetSupplierByAlgo(config, &algorithm)
 	//something went  wrong, TODO add error handler
 	if err != nil {
 		handleError("Can not create algorithm runner instance: "+err.Error(), w)
 		return
 	}
-	go s.Supplier.Run()
-	handleGet(s, w, r)
+	processId := uuid.New().String()
+	stat := supplier.Status()
+	stat.Type = status.STARTED
+	stat.Value = processId
+	jsonStat, err := json.Marshal(stat)
+	if err != nil {
+		handleError("Can not parse status"+err.Error(), w)
+		return
+	}
+	w.Write(jsonStat)
+	go supplier.Run()
+	s.suppliers[processId] = supplier
 	logger.Info("Process started")
+	s.ensureCapacity()
 }
 
 //terminates running calculation
-func handleDelete(s *OnPremiseServer, w http.ResponseWriter, r *http.Request) {
-	if s.Supplier == nil {
+func handleDelete(s *OnPremiseServer, w http.ResponseWriter, r *http.Request, id string) {
+	s.mux.Lock()
+	defer s.mux.Unlock()
+	supplier, ok := s.suppliers[id]
+	if !ok {
 		handleError("There is no running process", w)
 		return
 	}
-	s.Supplier.Terminate()
-	s.Supplier.Delete()
+
+	supplier.Terminate()
+	supplier.Delete()
 	w.WriteHeader(http.StatusOK)
-	s.Supplier = nil
+	delete(s.suppliers, id)
 	logger.Info("Process terminated")
 }
 
 //delegates GET POST DELETE main server listener
 func (s *OnPremiseServer) handle(w http.ResponseWriter, r *http.Request) {
-	s.mux.Lock()
-	defer s.mux.Unlock()
-
 	logger.LogRequest(r)
 	switch r.Method {
 	case "GET":
-		handleGet(s, w, r)
+		requestId := path.Base(html.EscapeString(r.URL.Path))
+		if !regex.MatchString(requestId) {
+			logger.Error("Request is not a valid request id! %s", requestId)
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+		handleGet(s, w, r, requestId)
 	case "POST":
 		handlePost(s, w, r)
 	case "DELETE":
-		handleDelete(s, w, r)
+		requestId := path.Base(html.EscapeString(r.URL.Path))
+		if !regex.MatchString(requestId) {
+			logger.Error("Request is not a valid request id! %s", requestId)
+			w.WriteHeader(http.StatusNotFound)
+			return
+		}
+
+		handleDelete(s, w, r, requestId)
 	default:
 		w.WriteHeader(http.StatusMethodNotAllowed)
 		logger.Error("Can not handle request method rejecting request")
@@ -147,3 +182,18 @@ func handleError(message string, w http.ResponseWriter, params ...interface{}) {
 	w.Write(jsonStatus)
 	return
 }
+
+func (s *OnPremiseServer) ensureCapacity() {
+	if len(s.suppliers) >= default_capacity {
+		var i int = 0
+		for k := range s.suppliers {
+			if i >= threshold {
+				supplier := s.suppliers[k]
+				if supplier.Status().Type >= status.RUNNING {
+					delete(s.suppliers, k)
+				}
+			}
+			i++
+		}
+	}
+}

internal/supplier/supplierfactory.go

@@ -35,8 +35,6 @@ func (factory *SupplierFactory) GetSupplierByAlgo(config *Config, algorithm *Alg
 
 }
 
-var cache map[string]Supplier = make(map[string]Supplier)
-
 func isSupplierSupportsAlgorithm(supplier Supplier, algo *Algorithm) bool {
 	supports := false
 	for _, supportedAlgo := range supplier.Supports() {
@@ -50,39 +48,21 @@ func isSupplierSupportsAlgorithm(supplier Supplier, algo *Algorithm) bool {
 //creates supplier instance
 func getSupplierInstance(config *Config, algo *Algorithm) (Supplier, error) {
 	if *config.Supplier == "go" {
-		s, ok := cache["go"+string(*algo)]
-		if !ok {
-			s = &GoSupplier{}
-			cache["go"+string(*algo)] = s
-
-		}
+		s := &GoSupplier{}
 		setFields(s.Data(), *algo, config)
 		return s, nil
 	} else if *config.Supplier == "openssl" {
-		s, ok := cache["openssl"+string(*algo)]
-		if !ok {
-			s = &CommandSupplier{Type: OPENSSL}
-			cache["openssl"+string(*algo)] = s
-
-		}
+		s := &CommandSupplier{Type: OPENSSL}
 		setFields(s.Data(), *algo, config)
 		return s, nil
 	}
 	switch runtime.GOOS {
 	case "linux", "mac":
-		s, ok := cache["mac"+string(*algo)]
-		if !ok {
-			s = &CommandSupplier{Type: UNIX}
-			cache["mac"+string(*algo)] = s
-		}
+		s := &CommandSupplier{Type: UNIX}
 		setFields(s.Data(), *algo, config)
 		return s, nil
 	case "windows":
-		s, ok := cache["windows"+string(*algo)]
-		if !ok {
-			s = &CommandSupplier{Type: WINDOWS}
-			cache["windows"+string(*algo)] = s
-		}
+		s := &CommandSupplier{Type: WINDOWS}
 		setFields(s.Data(), *algo, config)
 		return s, nil
 	default:
@@ -94,17 +74,13 @@ func getSupplierInstance(config *Config, algo *Algorithm) (Supplier, error) {
 
 //utility to set fields
 func setFields(base *BaseSupplier, algo Algorithm, config *Config) {
-	if base.status == nil {
-		base.status = &status.Status{}
-	}
+	base.status = &status.Status{}
 	base.status.Type = status.PREPARED
 	base.status.Value = ""
 	base.status.Checksum = ""
 	base.Algorithm = algo
 	base.Key = *config.Key
-	if base.File == nil {
-		base.File = &File{}
-	}
+	base.File = &File{}
 	base.File.Reset()
 	base.File.Url = *config.File
 	base.File.Status = base.status