Fix #657: Pass deadline for a task from task maniest or from command …

…line for a workflow manifest.
intelsdi-x · Jan 8, 2016 · 9e7c045 · 9e7c045
1 parent 05a376c
commit 9e7c045
Show file tree

Hide file tree

Showing 6 changed files with 29 additions and 9 deletions.
diff --git a/cmd/snapctl/commands.go b/cmd/snapctl/commands.go
@@ -48,6 +48,7 @@ var (
 						flTaskName,
 						flTaskSchedDuration,
 						flTaskSchedNoStart,
+						flTaskDeadline,
 					},
 				},
 				{

diff --git a/cmd/snapctl/flags.go b/cmd/snapctl/flags.go
@@ -108,6 +108,10 @@ var (
 		Name:  "no-start",
 		Usage: "Do not start task on creation [normally started on creation]",
 	}
+	flTaskDeadline = cli.StringFlag{
+		Name:  "deadline",
+		Usage: "The deadline for the task to be killed after started if the task runs too long (All tasks default to 5s)",
+	}
 
 	// metric
 	flMetricVersion = cli.IntFlag{

diff --git a/cmd/snapctl/task.go b/cmd/snapctl/task.go
@@ -88,6 +88,7 @@ type task struct {
 	Schedule *client.Schedule
 	Workflow *wmap.WorkflowMap
 	Name     string
+	Deadline string
 }
 
 func createTask(ctx *cli.Context) {
@@ -140,7 +141,7 @@ func createTaskUsingTaskManifest(ctx *cli.Context) {
 		fmt.Println("Invalid version provided")
 		os.Exit(1)
 	}
-	r := pClient.CreateTask(t.Schedule, t.Workflow, t.Name, !ctx.IsSet("no-start"))
+	r := pClient.CreateTask(t.Schedule, t.Workflow, t.Name, t.Deadline, !ctx.IsSet("no-start"))
 
 	if r.Err != nil {
 		errors := strings.Split(r.Err.Error(), " -- ")
@@ -194,6 +195,9 @@ func createTaskUsingWFManifest(ctx *cli.Context) {
 		os.Exit(1)
 	}
 
+	// Deadline for a task
+	dl := ctx.String("deadline")
+
 	var sch *client.Schedule
 	// None of these mean it is a simple schedule
 	if !ctx.IsSet("start-date") && !ctx.IsSet("start-time") && !ctx.IsSet("stop-date") && !ctx.IsSet("stop-time") {
@@ -256,7 +260,7 @@ func createTaskUsingWFManifest(ctx *cli.Context) {
 		}
 	}
 	// Create task
-	r := pClient.CreateTask(sch, wf, name, !ctx.IsSet("no-start"))
+	r := pClient.CreateTask(sch, wf, name, dl, !ctx.IsSet("no-start"))
 	if r.Err != nil {
 		errors := strings.Split(r.Err.Error(), " -- ")
 		fmt.Println("Error creating task:")

diff --git a/core/task.go b/core/task.go
@@ -98,6 +98,14 @@ func TaskDeadlineDuration(v time.Duration) TaskOption {
 	return func(t Task) TaskOption {
 		previous := t.DeadlineDuration()
 		t.SetDeadlineDuration(v)
+		log.WithFields(log.Fields{
+			"_module":                "core",
+			"_block":                 "TaskDeadlineDuration",
+			"task-id":                t.ID(),
+			"task-name":              t.GetName(),
+			"task deadline duration": t.DeadlineDuration(),
+		}).Debug("Setting deadlineDuration on task")
+
 		return TaskDeadlineDuration(previous)
 	}
 }

diff --git a/mgmt/rest/client/client_func_test.go b/mgmt/rest/client/client_func_test.go
@@ -146,7 +146,7 @@ func TestSnapClient(t *testing.T) {
 				So(t1.Err.Error(), ShouldEqual, fmt.Sprintf("Task not found: ID(%s)", uuid))
 			})
 			Convey("invalid task (missing metric)", func() {
-				tt := c.CreateTask(sch, wf, "baron", true)
+				tt := c.CreateTask(sch, wf, "baron", "", true)
 				So(tt.Err, ShouldNotBeNil)
 				So(tt.Err.Error(), ShouldContainSubstring, "Metric not found: /intel/mock/foo")
 			})
@@ -173,7 +173,7 @@ func TestSnapClient(t *testing.T) {
 			So(p.AvailablePlugins, ShouldBeEmpty)
 		})
 		Convey("invalid task (missing publisher)", func() {
-			tf := c.CreateTask(sch, wf, "baron", false)
+			tf := c.CreateTask(sch, wf, "baron", "", false)
 			So(tf.Err, ShouldNotBeNil)
 			So(tf.Err.Error(), ShouldContainSubstring, "Plugin not found: type(publisher) name(file)")
 		})
@@ -263,11 +263,11 @@ func TestSnapClient(t *testing.T) {
 	Convey("Tasks", t, func() {
 		Convey("Passing a bad task manifest", func() {
 			wfb := getWMFromSample("bad.json")
-			ttb := c.CreateTask(sch, wfb, "bad", true)
+			ttb := c.CreateTask(sch, wfb, "bad", "", true)
 			So(ttb.Err, ShouldNotBeNil)
 		})
 
-		tf := c.CreateTask(sch, wf, "baron", false)
+		tf := c.CreateTask(sch, wf, "baron", "", false)
 		Convey("valid task not started on creation", func() {
 			So(tf.Err, ShouldBeNil)
 			So(tf.Name, ShouldEqual, "baron")
@@ -310,7 +310,7 @@ func TestSnapClient(t *testing.T) {
 			})
 		})
 
-		tt := c.CreateTask(sch, wf, "baron", true)
+		tt := c.CreateTask(sch, wf, "baron", "", true)
 		Convey("valid task started on creation", func() {
 			So(tt.Err, ShouldBeNil)
 			So(tt.Name, ShouldEqual, "baron")
@@ -398,7 +398,7 @@ func TestSnapClient(t *testing.T) {
 				Convey("event stream", func() {
 					rest.StreamingBufferWindow = 0.01
 					sch := &Schedule{Type: "simple", Interval: "500ms"}
-					tf := c.CreateTask(sch, wf, "baron", false)
+					tf := c.CreateTask(sch, wf, "baron", "", false)
 
 					type ea struct {
 						events []string

diff --git a/mgmt/rest/client/task.go b/mgmt/rest/client/task.go
@@ -47,7 +47,7 @@ type Schedule struct {
 // If the startTask flag is true, the newly created task is started after the creation.
 // Otherwise, it's in the Stopped state. CreateTask is accomplished through a POST HTTP JSON request.
 // A ScheduledTask is returned if it succeeds, otherwise an error is returned.
-func (c *Client) CreateTask(s *Schedule, wf *wmap.WorkflowMap, name string, startTask bool) *CreateTaskResult {
+func (c *Client) CreateTask(s *Schedule, wf *wmap.WorkflowMap, name string, deadline string, startTask bool) *CreateTaskResult {
 	t := request.TaskCreationRequest{
 		Schedule: request.Schedule{
 			Type:     s.Type,
@@ -69,6 +69,9 @@ func (c *Client) CreateTask(s *Schedule, wf *wmap.WorkflowMap, name string, star
 	if name != "" {
 		t.Name = name
 	}
+	if deadline != "" {
+		t.Deadline = deadline
+	}
 	// Marshal to JSON for request body
 	j, err := json.Marshal(t)
 	if err != nil {