fixed issues

Author: tikazyq

controllers/task.go

@@ -107,6 +107,7 @@ func (ctx *taskContext) run(c *gin.Context) {
 	opts := &interfaces.SpiderRunOptions{
 		Mode:     t.Mode,
 		NodeIds:  t.NodeIds,
+		Cmd:      t.Cmd,
 		Param:    t.Param,
 		Priority: t.Priority,
 	}

grpc/server/message_server.go

@@ -28,27 +28,31 @@ func (svr MessageServer) Connect(stream grpc.MessageService_ConnectServer) (err
 	finished := make(chan bool)
 	for {
 		msg, err := stream.Recv()
+		nodeKey := "unknown node key"
+		if msg != nil {
+			nodeKey = msg.NodeKey
+		}
 		if err == io.EOF {
-			log.Infof("[MessageServer] received signal EOF from node[%s], now quit", msg.NodeKey)
+			log.Infof("[MessageServer] received signal EOF from node[%s], now quit", nodeKey)
 			return nil
 		}
 		if err != nil {
-			log.Errorf("[MessageServer] receiving message error from node[%s]: %v", msg.NodeKey, err)
+			log.Errorf("[MessageServer] receiving message error from node[%s]: %v", nodeKey, err)
 			return err
 		}
 		switch msg.Code {
 		case grpc.StreamMessageCode_CONNECT:
-			log.Infof("[MessageServer] received connect request from node[%s], key: %s", msg.NodeKey, msg.Key)
+			log.Infof("[MessageServer] received connect request from node[%s], key: %s", nodeKey, msg.Key)
 			svr.server.SetSubscribe(msg.Key, &entity.GrpcSubscribe{
 				Stream:   stream,
 				Finished: finished,
 			})
 		case grpc.StreamMessageCode_DISCONNECT:
-			log.Infof("[MessageServer] received disconnect request from node[%s], key: %s", msg.NodeKey, msg.Key)
+			log.Infof("[MessageServer] received disconnect request from node[%s], key: %s", nodeKey, msg.Key)
 			svr.server.DeleteSubscribe(msg.Key)
 			return nil
 		case grpc.StreamMessageCode_SEND:
-			log.Debugf("[MessageServer] received send request from node[%s] to %s", msg.NodeKey, msg.To)
+			log.Debugf("[MessageServer] received send request from node[%s] to %s", nodeKey, msg.To)
 			sub, err := svr.server.GetSubscribe(msg.To)
 			if err != nil {
 				return err

grpc/server/server.go

@@ -2,6 +2,7 @@ package server
 
 import (
 	"encoding/json"
+	"fmt"
 	"github.com/apex/log"
 	config2 "github.com/crawlab-team/crawlab-core/config"
 	"github.com/crawlab-team/crawlab-core/constants"
@@ -180,12 +181,13 @@ func (svr *Server) IsStopped() (res bool) {
 	return svr.stopped
 }
 
+func (svr *Server) recoveryHandlerFunc(p interface{}) (err error) {
+	err = errors.NewError(errors.ErrorPrefixGrpc, fmt.Sprintf("%v", p))
+	trace.PrintError(err)
+	return err
+}
+
 func NewServer(opts ...Option) (svr2 interfaces.GrpcServer, err error) {
-	// recovery options
-	var recoveryFunc grpc_recovery.RecoveryHandlerFunc
-	recoveryOpts := []grpc_recovery.Option{
-		grpc_recovery.WithRecoveryHandler(recoveryFunc),
-	}
 
 	// server
 	svr := &Server{
@@ -244,6 +246,11 @@ func NewServer(opts ...Option) (svr2 interfaces.GrpcServer, err error) {
 		return nil, err
 	}
 
+	// recovery options
+	recoveryOpts := []grpc_recovery.Option{
+		grpc_recovery.WithRecoveryHandler(svr.recoveryHandlerFunc),
+	}
+
 	// grpc server
 	svr.svr = grpc.NewServer(
 		grpc_middleware.WithUnaryServerChain(

spider/admin/service.go

@@ -71,6 +71,23 @@ func (svc *Service) scheduleTasks(s *models.Spider, opts *interfaces.SpiderRunOp
 		UserId:     opts.UserId,
 	}
 
+	// normalize
+	if mainTask.Mode == "" {
+		mainTask.Mode = s.Mode
+	}
+	if mainTask.NodeIds == nil {
+		mainTask.NodeIds = s.NodeIds
+	}
+	if mainTask.Cmd == "" {
+		mainTask.Cmd = s.Cmd
+	}
+	if mainTask.Param == "" {
+		mainTask.Param = s.Param
+	}
+	if mainTask.Priority == 0 {
+		mainTask.Priority = s.Priority
+	}
+
 	if svc.isMultiTask(opts) {
 		// multi tasks
 		// TODO: implement associated tasks