Merge branch 'release/2.2.0' of ssh://git.yunion.io/~qiujian/onecloud into hotfix/qj-resolve-conflict-20181022-2.3.0

Conflicts:
	pkg/httperrors/httperrors.go

pkg/appsrv/appsrv.go

@@ -10,83 +10,19 @@ import (
 	"time"
 
 	"yunion.io/x/log"
-	"yunion.io/x/onecloud/pkg/appctx"
-	"yunion.io/x/onecloud/pkg/proxy"
 	"yunion.io/x/pkg/trace"
 	"yunion.io/x/pkg/utils"
+
+	"yunion.io/x/onecloud/pkg/appctx"
+	"yunion.io/x/onecloud/pkg/httperrors"
+	"yunion.io/x/onecloud/pkg/proxy"
+	"yunion.io/x/onecloud/pkg/util/httputils"
 )
 
-type responseWriterResponse struct {
-	count int
-	err   error
-}
-
-type responseWriterChannel struct {
-	backend    http.ResponseWriter
-	bodyChan   chan []byte
-	bodyResp   chan responseWriterResponse
-	statusChan chan int
-	statusResp chan bool
-}
-
-func newResponseWriterChannel(backend http.ResponseWriter) responseWriterChannel {
-	return responseWriterChannel{backend: backend,
-		bodyChan:   make(chan []byte),
-		bodyResp:   make(chan responseWriterResponse),
-		statusChan: make(chan int),
-		statusResp: make(chan bool)}
-}
-
-func (w *responseWriterChannel) Header() http.Header {
-	return w.backend.Header()
-}
-
-func (w *responseWriterChannel) Write(bytes []byte) (int, error) {
-	w.bodyChan <- bytes
-	v := <-w.bodyResp
-	return v.count, v.err
-}
-
-func (w *responseWriterChannel) WriteHeader(status int) {
-	w.statusChan <- status
-	<-w.statusResp
-}
-
-func (w *responseWriterChannel) wait() {
-	stop := false
-	for !stop {
-		select {
-		case bytes, more := <-w.bodyChan:
-			// log.Print("Recive body ", len(bytes), " more ", more)
-			if more {
-				c, e := w.backend.Write(bytes)
-				w.bodyResp <- responseWriterResponse{count: c, err: e}
-			} else {
-				stop = true
-			}
-		case status, more := <-w.statusChan:
-			// log.Print("Recive status ", status, " more ", more)
-			if more {
-				w.backend.WriteHeader(status)
-				w.statusResp <- true
-			} else {
-				stop = true
-			}
-		}
-	}
-}
-
-func (w *responseWriterChannel) closeChannels() {
-	close(w.bodyChan)
-	close(w.bodyResp)
-	close(w.statusChan)
-	close(w.statusResp)
-}
-
 type Application struct {
 	name              string
 	context           context.Context
-	session           *WorkerManager
+	session           *SWorkerManager
 	roots             map[string]*RadixNode
 	rootLock          *sync.Mutex
 	connMax           int
@@ -106,19 +42,22 @@ const (
 	DEFAULT_READ_TIMEOUT        = 0
 	DEFAULT_READ_HEADER_TIMEOUT = 10 * time.Second
 	DEFAULT_WRITE_TIMEOUT       = 0
+	DEFAULT_PROCESS_TIMEOUT     = 15 * time.Second
 )
 
 func NewApplication(name string, connMax int) *Application {
 	app := Application{name: name,
 		context:           context.Background(),
 		connMax:           connMax,
-		session:           NewWorkerManager("sessionMan", connMax, DEFAULT_BACKLOG),
+		session:           NewWorkerManager("HttpRequestWorkerManager", connMax, DEFAULT_BACKLOG),
 		roots:             make(map[string]*RadixNode),
 		rootLock:          &sync.Mutex{},
 		idleTimeout:       DEFAULT_IDLE_TIMEOUT,
 		readTimeout:       DEFAULT_READ_TIMEOUT,
 		readHeaderTimeout: DEFAULT_READ_HEADER_TIMEOUT,
-		writeTimeout:      DEFAULT_WRITE_TIMEOUT}
+		writeTimeout:      DEFAULT_WRITE_TIMEOUT,
+		processTimeout:    DEFAULT_PROCESS_TIMEOUT,
+	}
 	app.SetContext(appctx.APP_CONTEXT_KEY_APP, &app)
 	app.SetContext(appctx.APP_CONTEXT_KEY_APPNAME, app.name)
 
@@ -174,9 +113,6 @@ func (app *Application) AddHandler(method string, prefix string, handler func(co
 func (app *Application) AddHandler2(method string, prefix string, handler func(context.Context, http.ResponseWriter, *http.Request), metadata map[string]interface{}, name string, tags map[string]string) {
 	log.Debugf("%s - %s", method, prefix)
 	segs := SplitPath(prefix)
-	// for i := len(this.middlewares) - 1; i >= 0; i -= 1 {
-	// 	handler = this.middlewares[i](handler)
-	// }
 	e := app.getRoot(method).Add(segs, newHandlerInfo(method, segs, handler, metadata, name, tags))
 	if e != nil {
 		log.Fatalf("Fail to register %s %s: %s", method, prefix, e)
@@ -253,10 +189,11 @@ func (app *Application) defaultHandle(w http.ResponseWriter, r *http.Request, ri
 		hand, ok := handler.(*handlerInfo)
 		if ok {
 			fw := newResponseWriterChannel(w)
+			worker := make(chan *SWorker)
 			errChan := make(chan interface{})
+			ctx, cancel := context.WithTimeout(app.context, app.processTimeout)
+			defer cancel()
 			app.session.Run(func() {
-				ctx, cancel := context.WithCancel(app.context)
-				defer cancel()
 				defer fw.closeChannels()
 				if ctx.Err() == nil {
 					ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_REQUEST_ID, rid)
@@ -266,25 +203,32 @@ func (app *Application) defaultHandle(w http.ResponseWriter, r *http.Request, ri
 					if hand.metadata != nil {
 						ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_METADATA, hand.metadata)
 					}
-					span := trace.StartServerTrace(w, r, hand.GetName(params), app.GetName(), hand.GetTags())
-					ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_TRACE, span)
-					hand.handler(ctx, &fw, r)
-					span.EndTrace()
+					func() {
+						span := trace.StartServerTrace(w, r, hand.GetName(params), app.GetName(), hand.GetTags())
+						defer span.EndTrace()
+						ctx = context.WithValue(ctx, appctx.APP_CONTEXT_KEY_TRACE, span)
+						hand.handler(ctx, &fw, r)
+					}()
 				} // otherwise, the task has been timeout
-			}, errChan)
-			fw.wait()
-			runerr := WaitChannel(errChan)
-			if runerr != nil {
-				http.Error(w, fmt.Sprintf("Internal error: %s", runerr), http.StatusInternalServerError)
+			}, worker, errChan)
+			runErr := fw.wait(ctx, worker, errChan)
+			if runErr != nil {
+				switch runErr.(type) {
+				case *httputils.JSONClientError:
+					je := runErr.(*httputils.JSONClientError)
+					httperrors.GeneralServerError(w, je)
+				default:
+					httperrors.InternalServerError(w, "Internal server error")
+				}
 			}
 			return hand
 		} else {
-			log.Printf("Invalid handler for %s", r.URL)
-			http.Error(w, "Invalid handler", 500)
+			log.Errorf("Invalid handler for %s", r.URL)
+			httperrors.InternalServerError(w, "Invalid handler %s", r.URL)
 		}
 	} else if !isCors {
-		log.Printf("Handler not found")
-		http.NotFound(w, r)
+		log.Errorf("Handler not found")
+		httperrors.NotFoundError(w, "Handler not found")
 	}
 	return nil
 }
@@ -295,6 +239,7 @@ func (app *Application) addDefaultHandler() {
 	app.AddHandler("POST", "/ping", PingHandler)
 	app.AddHandler("GET", "/ping", PingHandler)
 	// app.AddHandler("OPTIONS", "/", CORSHandler)
+	app.AddHandler("GET", "/worker_stats", WorkerStatsHandler)
 }
 
 func timeoutHandle(h http.Handler) http.HandlerFunc {

pkg/appsrv/response.go

@@ -0,0 +1,95 @@
+package appsrv
+
+import (
+	"context"
+	"net/http"
+
+	"yunion.io/x/log"
+
+	"yunion.io/x/onecloud/pkg/httperrors"
+)
+
+type responseWriterResponse struct {
+	count int
+	err   error
+}
+
+type responseWriterChannel struct {
+	backend    http.ResponseWriter
+	bodyChan   chan []byte
+	bodyResp   chan responseWriterResponse
+	statusChan chan int
+	statusResp chan bool
+}
+
+func newResponseWriterChannel(backend http.ResponseWriter) responseWriterChannel {
+	return responseWriterChannel{backend: backend,
+		bodyChan:   make(chan []byte),
+		bodyResp:   make(chan responseWriterResponse),
+		statusChan: make(chan int),
+		statusResp: make(chan bool)}
+}
+
+func (w *responseWriterChannel) Header() http.Header {
+	return w.backend.Header()
+}
+
+func (w *responseWriterChannel) Write(bytes []byte) (int, error) {
+	w.bodyChan <- bytes
+	v := <-w.bodyResp
+	return v.count, v.err
+}
+
+func (w *responseWriterChannel) WriteHeader(status int) {
+	w.statusChan <- status
+	<-w.statusResp
+}
+
+func (w *responseWriterChannel) wait(ctx context.Context, workerChan chan *SWorker, errChan chan interface{}) interface{} {
+	var err interface{}
+	var worker *SWorker
+	stop := false
+	for !stop {
+		select {
+		case worker = <-workerChan:
+			log.Infof("request is being handled by worker %s", worker)
+		case <-ctx.Done():
+			// ctx deadline reached, timeout
+			if worker != nil {
+				worker.Detach("timeout")
+			}
+			err = httperrors.NewTimeoutError("request process timeout")
+			stop = true
+		case e, more := <-errChan:
+			if more {
+				err = e
+			} else {
+				stop = true
+			}
+		case bytes, more := <-w.bodyChan:
+			// log.Print("Recive body ", len(bytes), " more ", more)
+			if more {
+				c, e := w.backend.Write(bytes)
+				w.bodyResp <- responseWriterResponse{count: c, err: e}
+			} else {
+				stop = true
+			}
+		case status, more := <-w.statusChan:
+			// log.Print("Recive status ", status, " more ", more)
+			if more {
+				w.backend.WriteHeader(status)
+				w.statusResp <- true
+			} else {
+				stop = true
+			}
+		}
+	}
+	return err
+}
+
+func (w *responseWriterChannel) closeChannels() {
+	close(w.bodyChan)
+	close(w.bodyResp)
+	close(w.statusChan)
+	close(w.statusResp)
+}

pkg/appsrv/workers.go

@@ -1,64 +1,178 @@
 package appsrv
 
 import (
+	"container/list"
+	"context"
+	"fmt"
+	"net/http"
 	"runtime/debug"
 	"sync"
 
+	"yunion.io/x/jsonutils"
 	"yunion.io/x/log"
 )
 
-type WorkerManager struct {
-	name         string
-	queue        *Ring
-	workerCount  int
-	backlog      int
-	activeWorker int
-	workerLock   *sync.Mutex
-	workerId     uint64
+const (
+	WORKER_STATE_ACTIVE = 0
+	WORKER_STATE_DETACH = 1
+)
+
+var isDebug = false
+
+func enableDebug() {
+	isDebug = true
 }
 
-func NewWorkerManager(name string, workerCount int, backlog int) *WorkerManager {
-	manager := WorkerManager{name: name,
-		queue:        NewRing(workerCount * backlog),
-		workerCount:  workerCount,
-		backlog:      backlog,
-		activeWorker: 0,
-		workerLock:   &sync.Mutex{},
-		workerId:     0}
+var workerManagers []*SWorkerManager
+
+func init() {
+	workerManagers = make([]*SWorkerManager, 0)
+}
+
+type SWorker struct {
+	id        uint64
+	state     int
+	container *list.Element
+	manager   *SWorkerManager
+}
+
+func newWorker(id uint64, manager *SWorkerManager) *SWorker {
+	return &SWorker{
+		id:        id,
+		state:     WORKER_STATE_ACTIVE,
+		container: nil,
+		manager:   manager,
+	}
+}
+
+func (worker *SWorker) isDetached() bool {
+	worker.manager.workerLock.Lock()
+	defer worker.manager.workerLock.Unlock()
+
+	return worker.state == WORKER_STATE_DETACH
+}
+
+func (worker *SWorker) run() {
+	for {
+		if worker.isDetached() {
+			if isDebug {
+				log.Debugf("deteched worker %s, no need to pick up new job", worker)
+			}
+			break
+		}
+		req := worker.manager.queue.Pop()
+		if req != nil {
+			task := req.(*sWorkerTask)
+			if task.worker != nil {
+				task.worker <- worker
+			}
+			if isDebug {
+				log.Debugf("start exec task on worker %s", worker)
+			}
+			execCallback(task)
+			if isDebug {
+				log.Debugf("end exec task on worker %s", worker)
+			}
+		} else {
+			if isDebug {
+				log.Debugf("no more job, exit worker %s", worker)
+			}
+			break
+		}
+	}
+	worker.manager.removeWorker(worker)
+}
+
+func (worker *SWorker) Detach(reason string) {
+	worker.manager.workerLock.Lock()
+	defer worker.manager.workerLock.Unlock()
+
+	worker.state = WORKER_STATE_DETACH
+	worker.manager.activeWorker.removeWithLock(worker)
+	worker.manager.detachedWorker.addWithLock(worker)
+
+	log.Warningf("detach worker %s due to reason %s", worker, reason)
+}
+
+func (worker *SWorker) String() string {
+	return fmt.Sprintf("#%d(%d)", worker.id, worker.state)
+}
+
+type SWorkerList struct {
+	list *list.List
+}
+
+func newWorkerList() SWorkerList {
+	return SWorkerList{
+		list: list.New(),
+	}
+}
+
+func (wl *SWorkerList) addWithLock(worker *SWorker) {
+	ele := wl.list.PushBack(worker)
+	worker.container = ele
+}
+
+func (wl *SWorkerList) removeWithLock(worker *SWorker) {
+	wl.list.Remove(worker.container)
+	worker.container = nil
+}
+
+func (wl *SWorkerList) size() int {
+	return wl.list.Len()
+}
+
+type SWorkerManager struct {
+	name           string
+	queue          *Ring
+	workerCount    int
+	backlog        int
+	activeWorker   SWorkerList
+	detachedWorker SWorkerList
+	workerLock     *sync.Mutex
+	workerId       uint64
+}
+
+func NewWorkerManager(name string, workerCount int, backlog int) *SWorkerManager {
+	manager := SWorkerManager{name: name,
+		queue:          NewRing(workerCount * backlog),
+		workerCount:    workerCount,
+		backlog:        backlog,
+		activeWorker:   newWorkerList(),
+		detachedWorker: newWorkerList(),
+		workerLock:     &sync.Mutex{},
+		workerId:       0}
+
+	workerManagers = append(workerManagers, &manager)
 	return &manager
 }
 
-type workerTask struct {
-	task func()
-	err  chan interface{}
+type sWorkerTask struct {
+	task   func()
+	worker chan *SWorker
+	err    chan interface{}
 }
 
-func (wm *WorkerManager) Run(task func(), err chan interface{}) bool {
-	ret := wm.queue.Push(&workerTask{task: task, err: err})
+func (wm *SWorkerManager) Run(task func(), worker chan *SWorker, err chan interface{}) bool {
+	ret := wm.queue.Push(&sWorkerTask{task: task, worker: worker, err: err})
 	if ret {
 		wm.schedule()
 	}
 	return ret
 }
 
-func (wm *WorkerManager) workerRun(id uint64) {
-	//log.Println("Start worker", id)
-	defer wm.decActiveWorker()
-	req := wm.queue.Pop()
-	for req != nil {
-		wm.execCallback(req.(*workerTask))
-		req = wm.queue.Pop()
-	}
-	//log.Println("End worker", id)
-}
-
-func (wm *WorkerManager) decActiveWorker() {
+func (wm *SWorkerManager) removeWorker(worker *SWorker) {
 	wm.workerLock.Lock()
 	defer wm.workerLock.Unlock()
-	wm.activeWorker -= 1
+
+	if worker.state == WORKER_STATE_ACTIVE {
+		wm.activeWorker.removeWithLock(worker)
+	} else {
+		wm.detachedWorker.removeWithLock(worker)
+	}
 }
 
-func (wm *WorkerManager) execCallback(task *workerTask) {
+func execCallback(task *sWorkerTask) {
 	defer func() {
 		if r := recover(); r != nil {
 			log.Errorf("WorkerManager exec callback error: %s", r)
@@ -75,16 +189,59 @@ func (wm *WorkerManager) execCallback(task *workerTask) {
 	}
 }
 
-func (wm *WorkerManager) schedule() {
+func (wm *SWorkerManager) schedule() {
 	wm.workerLock.Lock()
 	defer wm.workerLock.Unlock()
-	if wm.activeWorker < wm.workerCount && wm.queue.Size() > 0 {
-		wm.activeWorker += 1
+
+	if wm.activeWorker.size() < wm.workerCount && wm.queue.Size() > 0 {
 		wm.workerId += 1
-		go wm.workerRun(wm.workerId)
+		worker := newWorker(wm.workerId, wm)
+		wm.activeWorker.addWithLock(worker)
+		if isDebug {
+			log.Debugf("no enough worker, add new worker %s", worker)
+		}
+		go worker.run()
 	}
 }
 
+func (wm *SWorkerManager) ActiveWorkerCount() int {
+	return wm.activeWorker.size()
+}
+
+func (wm *SWorkerManager) DetachedWorkerCount() int {
+	return wm.detachedWorker.size()
+}
+
+type SWorkerManagerStates struct {
+	Name            string
+	Backlog         int
+	MaxWorkerCnt    int
+	ActiveWorkerCnt int
+	DetachWorkerCnt int
+}
+
+func (wm *SWorkerManager) getState() SWorkerManagerStates {
+	state := SWorkerManagerStates{}
+
+	state.Name = wm.name
+	state.Backlog = wm.queue.Size()
+	state.MaxWorkerCnt = wm.workerCount
+	state.ActiveWorkerCnt = wm.activeWorker.size()
+	state.DetachWorkerCnt = wm.detachedWorker.size()
+
+	return state
+}
+
+func WorkerStatsHandler(ctx context.Context, w http.ResponseWriter, r *http.Request) {
+	stats := make([]SWorkerManagerStates, 0)
+	for i := 0; i < len(workerManagers); i += 1 {
+		stats = append(stats, workerManagers[i].getState())
+	}
+	result := jsonutils.NewDict()
+	result.Add(jsonutils.Marshal(&stats), "workers")
+	fmt.Fprintf(w, result.String())
+}
+
 func WaitChannel(ch chan interface{}) interface{} {
 	var ret interface{}
 	stop := false

pkg/appsrv/workers_test.go

@@ -6,22 +6,22 @@ import (
 )
 
 func TestWorkerManager(t *testing.T) {
+	enableDebug()
 	startTime := time.Now()
-	end := make(chan int)
+	// end := make(chan int)
 	wm := NewWorkerManager("testwm", 2, 10)
 	counter := 0
 	for i := 0; i < 10; i += 1 {
 		wm.Run(func() {
 			counter += 1
 			time.Sleep(1 * time.Second)
-			if counter >= i {
-				end <- 1
-			}
-		}, nil)
+		}, nil, nil)
+	}
+	for wm.ActiveWorkerCount() != 0 {
+		time.Sleep(time.Second)
 	}
-	<-end
 	if time.Since(startTime) < 5*time.Second {
-		t.Error("Increct timing")
+		t.Error("Incorrect timing")
 	}
 }
 
@@ -30,7 +30,7 @@ func TestWorkerManagerError(t *testing.T) {
 	err := make(chan interface{})
 	wm.Run(func() {
 		panic("Panic inside worker")
-	}, err)
+	}, nil, err)
 	e := WaitChannel(err)
 	if e == nil {
 		t.Error("Panic not captured")
@@ -38,9 +38,10 @@ func TestWorkerManagerError(t *testing.T) {
 	err = make(chan interface{})
 	wm.Run(func() {
 		time.Sleep(1 * time.Second)
-	}, err)
+	}, nil, err)
 	e = WaitChannel(err)
 	if e != nil {
 		t.Error("Should no error")
 	}
+
 }

pkg/cloudcommon/db/taskman/handler.go

@@ -8,7 +8,7 @@ import (
 	"yunion.io/x/onecloud/pkg/cloudcommon/db"
 )
 
-var taskWorkMan *appsrv.WorkerManager
+var taskWorkMan *appsrv.SWorkerManager
 
 func init() {
 	taskWorkMan = appsrv.NewWorkerManager("TaskWorkerManager", 4, 100)
@@ -22,5 +22,5 @@ func AddTaskHandler(prefix string, app *appsrv.Application) {
 func runTask(taskId string, data jsonutils.JSONObject) {
 	taskWorkMan.Run(func() {
 		TaskManager.execTask(taskId, data)
-	}, nil)
+	}, nil, nil)
 }

pkg/cloudcommon/db/taskman/localtaskworker.go

@@ -9,7 +9,7 @@ import (
 	"yunion.io/x/onecloud/pkg/appsrv"
 )
 
-var localTaskWorkerMan *appsrv.WorkerManager
+var localTaskWorkerMan *appsrv.SWorkerManager
 
 func init() {
 	localTaskWorkerMan = appsrv.NewWorkerManager("LocalTaskWorkerManager", 4, 10)
@@ -42,5 +42,5 @@ func LocalTaskRun(task ITask, proc func() (jsonutils.JSONObject, error)) {
 			task.ScheduleRun(data)
 		}
 
-	}, nil)
+	}, nil, nil)
 }

pkg/httperrors/errors.go

@@ -185,6 +185,11 @@ func NewRequireLicenseError(msg string, params ...interface{}) *httputils.JSONCl
 	return NewJsonClientError(402, "RequireLicenseError", msg, err)
 }
 
+func NewTimeoutError(msg string, params ...interface{}) *httputils.JSONClientError {
+	msg, err := errorMessage(msg, params...)
+	return NewJsonClientError(504, "TimeoutError", msg, err)
+}
+
 func NewGeneralError(err error) *httputils.JSONClientError {
 	switch err.(type) {
 	case *httputils.JSONClientError:

pkg/httperrors/httperrors.go

@@ -106,3 +106,7 @@ func NotSufficientPrivilegeError(w http.ResponseWriter, msg string, params ...in
 func ResourceNotFoundError(w http.ResponseWriter, msg string, params ...interface{}) {
 	JsonClientError(w, NewResourceNotFoundError(msg, params...))
 }
+
+func TimeoutError(w http.ResponseWriter, msg string, params ...interface{}) {
+	JsonClientError(w, NewTimeoutError(msg, params...))
+}

pkg/util/logclient/logclient.go

@@ -60,7 +60,7 @@ const (
 // golang 不支持 const 的string array, http://t.cn/EzAvbw8
 var BLACK_LIST_OBJ_TYPE = []string{"parameter"}
 
-var logclientWorkerMan *appsrv.WorkerManager
+var logclientWorkerMan *appsrv.SWorkerManager
 
 func init() {
 	logclientWorkerMan = appsrv.NewWorkerManager("LogClientWorkerManager", 1, 50)
@@ -119,5 +119,5 @@ func AddActionLog(model IObject, action string, iNotes interface{}, userCred mcc
 		if err != nil {
 			log.Errorf("create action log failed %s", err)
 		}
-	}, nil)
+	}, nil, nil)
 }