Feature/tag executions

CHANGELOG.md

@@ -15,6 +15,8 @@
 - (#316) Delete service when stoping the `service dev` command to avoid to keep all the versions of the services.
 - (#317) Add errors when trying to execute a service that is not running (nothing was happening before)
 - (#344) Add `service execute --data` flag to pass arguments as key=value.
+- (#362) Add `tags` list parameter for execution in order to be able to categorize and/or track a specific task execution.
+- (#362) Add possibility to listen results with specific tag(s)
 
 #### Removed
 - (#303) Deprecate command `service test` in favor of `service dev` and `service execute`

api/core/api.proto

@@ -79,13 +79,15 @@ message EventData {
 // {
 //   "serviceID":     "__SERVICE_ID__",
 //   "taskFilter":    "__TASK_KEY_TO_MATCH__",
-//   "outputFilter":  "__OUTPUT_KEY_TO_MATCH__"
+//   "outputFilter":  "__OUTPUT_KEY_TO_MATCH__",
+//   "tagFilters":     ["tagX"]
 // }
 // ```
 message ListenResultRequest {
-  string serviceID = 1;     // The Service ID. Generated when using the [`DeployService` API](#deployservice).
-  string taskFilter = 2;    // __Optional.__  The task's key to filter. The task must match this key. The default is `*` which matches any task.
-  string outputFilter = 3;  // __Optional.__ The output's key from the task to filter. The task must return this output's key. The default is `*` which matches any output.
+  string serviceID = 1;           // The Service ID. Generated when using the [`DeployService` API](#deployservice).
+  string taskFilter = 2;          // __Optional.__  The task's key to filter. The task must match this key. The default is `*` which matches any task.
+  string outputFilter = 3;        // __Optional.__ The output's key from the task to filter. The task must return this output's key. The default is `*` which matches any output.
+  repeated string tagFilters = 4; // __Optional.__ List of tags to filter. This is a "match all" list. All tags in parameters should be included in the execution to match.
 }
 
 // The data received from the stream of the `ListenResult` API.
@@ -97,14 +99,16 @@ message ListenResultRequest {
 //   "executionID": "__EXECUTION_ID__",
 //   "taskKey":     "__TASK_KEY__",
 //   "outputKey":   "__OUTPUT_KEY__",
-//   "outputData":  "{\"foo\":\"bar\"}"
+//   "outputData":  "{\"foo\":\"bar\"}",
+//   "tags":        ["executionX", "test"]
 // }
 // ```
 message ResultData {
-  string executionID = 1; // The unique identifier of the execution.
-  string taskKey = 2;     // The key of the executed task.
-  string outputKey = 3;   // The output's key from the returned task.
-  string outputData = 4;  // The output's data from the returned task, encoded in JSON.
+  string executionID = 1;   // The unique identifier of the execution.
+  string taskKey = 2;       // The key of the executed task.
+  string outputKey = 3;     // The output's key from the returned task.
+  string outputData = 4;    // The output's data from the returned task, encoded in JSON.
+  repeated string tags = 5; // The list of tags associated to the execution
 }
 
 // The request's data for the `ExecuteTask` API.
@@ -114,13 +118,15 @@ message ResultData {
 // {
 //   "serviceID": "__SERVICE_ID__",
 //   "taskKey":   "__TASK_KEY__",
-//   "inputData": "{\"foo\":\"bar\"}"
+//   "inputData": "{\"foo\":\"bar\"}",
+//   "tags":      ["executionX", "test"]
 // }
 // ```
 message ExecuteTaskRequest {
-  string serviceID = 1; // The Service ID. Generated when using the [`DeployService` API](#deployservice).
-  string taskKey = 2;   // The task's key to execute.
-  string inputData = 3; // The inputs of the task to execute, encoded in JSON.
+  string serviceID = 1;     // The Service ID. Generated when using the [`DeployService` API](#deployservice).
+  string taskKey = 2;       // The task's key to execute.
+  string inputData = 3;     // The inputs of the task to execute, encoded in JSON.
+  repeated string tags = 4; // The list of tags to associate to the execution
 }
 
 // The reply's data of the `ExecuteTask` API.

api/core/execute.go

@@ -22,7 +22,7 @@ func (s *Server) ExecuteTask(ctx context.Context, request *ExecuteTaskRequest) (
 	if err := checkServiceStatus(&srv); err != nil {
 		return nil, err
 	}
-	executionID, err := execute(&srv, request.TaskKey, inputs)
+	executionID, err := execute(&srv, request.TaskKey, inputs, request.Tags)
 	return &ExecuteTaskReply{
 		ExecutionID: executionID,
 	}, err
@@ -45,8 +45,8 @@ func getData(request *ExecuteTaskRequest) (map[string]interface{}, error) {
 	return inputs, err
 }
 
-func execute(srv *service.Service, key string, inputs map[string]interface{}) (executionID string, err error) {
-	exc, err := execution.Create(srv, key, inputs)
+func execute(srv *service.Service, key string, inputs map[string]interface{}, tags []string) (executionID string, err error) {
+	exc, err := execution.Create(srv, key, inputs, tags)
 	if err != nil {
 		return "", err
 	}

api/core/execute_test.go

@@ -126,14 +126,14 @@ func TestExecuteFunc(t *testing.T) {
 			"test": {},
 		},
 	}
-	id, err := execute(srv, "test", map[string]interface{}{})
+	id, err := execute(srv, "test", map[string]interface{}{}, []string{})
 	assert.Nil(t, err)
 	assert.NotNil(t, id)
 }
 
 func TestExecuteFuncInvalidTaskName(t *testing.T) {
 	srv := &service.Service{}
-	_, err := execute(srv, "test", map[string]interface{}{})
+	_, err := execute(srv, "test", map[string]interface{}{}, []string{})
 	assert.NotNil(t, err)
 }
 

api/core/listen_result.go

@@ -32,13 +32,14 @@ func (s *Server) ListenResult(request *ListenResultRequest, stream Core_ListenRe
 			return ctx.Err()
 		case data := <-subscription:
 			execution := data.(*execution.Execution)
-			if isSubscribedTask(request, execution) && isSubscribedOutput(request, execution) {
+			if isSubscribedToTags(request, execution) && isSubscribedToTask(request, execution) && isSubscribedToOutput(request, execution) {
 				outputs, _ := json.Marshal(execution.OutputData)
 				if err := stream.Send(&ResultData{
 					ExecutionID: execution.ID,
 					TaskKey:     execution.Task,
 					OutputKey:   execution.Output,
 					OutputData:  string(outputs),
+					Tags:        execution.Tags,
 				}); err != nil {
 					return err
 				}
@@ -79,10 +80,22 @@ func validateOutputKey(service *service.Service, taskKey string, outputFilter st
 	return nil
 }
 
-func isSubscribedTask(request *ListenResultRequest, e *execution.Execution) bool {
+func isSubscribedToTask(request *ListenResultRequest, e *execution.Execution) bool {
 	return array.IncludedIn([]string{"", "*", e.Task}, request.TaskFilter)
 }
 
-func isSubscribedOutput(request *ListenResultRequest, e *execution.Execution) bool {
+func isSubscribedToOutput(request *ListenResultRequest, e *execution.Execution) bool {
 	return array.IncludedIn([]string{"", "*", e.Output}, request.OutputFilter)
 }
+
+func isSubscribedToTags(request *ListenResultRequest, e *execution.Execution) bool {
+	if len(request.TagFilters) == 0 {
+		return true
+	}
+	for _, tag := range request.TagFilters {
+		if !array.IncludedIn(e.Tags, tag) {
+			return false
+		}
+	}
+	return true
+}