[OCTRL-683] Allow the OCClib-controlled tasks to exit on their own

knopers8 · teo · commit f81c9f7925f7 · 2022-08-11T10:38:46.000+02:00
Two problems are solved here:
- OccServer::EventStream thread would not exit when reaching down,
  thus making OCClib-controlled tasks never shut their server down
- executor would send SIGTERM as soon as a task reported reaching
  DONE, thus not allowing it to exit on its own.
diff --git a/core/integration/kafka/protos/kafka.proto b/core/integration/kafka/protos/kafka.proto
@@ -19,4 +19,4 @@ message EnvInfo {
   string state = 4;
   repeated string detectors = 5;
   uint64 enterStateTimestamp = 6; // ms since epoch.
-}
+}
diff --git a/executor/executable/controllabletask.go b/executor/executable/controllabletask.go
@@ -49,6 +49,7 @@ import (
 )
 
 const (
+	DONE_TIMEOUT            = 1 * time.Second
 	SIGTERM_TIMEOUT         = 1 * time.Second
 	SIGINT_TIMEOUT          = 3 * time.Second
 	KILL_TRANSITION_TIMEOUT = 1 * time.Second
@@ -417,13 +418,13 @@ func (t *ControllableTask) Launch() error {
 			}
 			for {
 				if t.rpc == nil {
-					log.WithField("partition", t.knownEnvironmentId.String()).
+					log.WithField("partition", t.knownEnvironmentId.String()).WithField("taskId", deo.TaskId.String()).
 						WithError(err).Debug("event stream done")
 					break
 				}
 				esr, err := esc.Recv()
 				if err == io.EOF {
-					log.WithField("partition", t.knownEnvironmentId.String()).
+					log.WithField("partition", t.knownEnvironmentId.String()).WithField("taskId", deo.TaskId.String()).
 						WithError(err).Debug("event stream EOF")
 					break
 				}
@@ -432,7 +433,7 @@ func (t *ControllableTask) Launch() error {
 						WithField("partition", t.knownEnvironmentId.String()).
 						WithField("errorType", reflect.TypeOf(err)).
 						WithField("level", infologger.IL_Devel).
-						Warning("error receiving event from task")
+						Warningf("error receiving event from task %s", deo.TaskId.String())
 					if status.Code(err) == codes.Unavailable {
 						break
 					}
@@ -690,15 +691,20 @@ func (t *ControllableTask) Kill() error {
 	if reachedState == "DONE" {
 		log.WithField("partition", t.knownEnvironmentId.String()).
 			WithField("taskId", t.ti.TaskID.Value).
-			Debug("task exited correctly")
+			Debug("task reached DONE, will try to terminate it gently")
 		t.pendingFinalTaskStateCh <- mesos.TASK_FINISHED
 	} else { // something went wrong
 		log.WithField("partition", t.knownEnvironmentId.String()).
 			WithField("taskId", t.ti.TaskID.Value).
-			Debug("task killed")
+			Debug("task already died or will be killed soon")
 		t.pendingFinalTaskStateCh <- mesos.TASK_KILLED
 	}
 
+	if reachedState == "DONE" {
+		log.WithField("taskId", t.ti.GetTaskID()).Debugf("waiting %.1fs before sending SIGTERM to task", DONE_TIMEOUT.Seconds())
+		time.Sleep(DONE_TIMEOUT)
+	}
+
 	return t.doTermIntKill(pid)
 }
 
diff --git a/occ/occlib/OccServer.cxx b/occ/occlib/OccServer.cxx
@@ -75,7 +75,7 @@ grpc::Status OccServer::EventStream(grpc::ServerContext* context,
     });
 
     bool isStreamOpen = true;
-    while (!m_destroying && isStreamOpen) {
+    while (!m_destroying && isStreamOpen && m_rco->getState() != t_State::done) {
         pb::DeviceEvent *newEvent;
         bool ok = eventQueue.pop(newEvent);
         if (!ok) {  // queue empty, sleep and retry
diff --git a/occ/plugin/OccLiteServer.cxx b/occ/plugin/OccLiteServer.cxx
@@ -106,7 +106,7 @@ ::grpc::Status OccLite::Service::Transition(::grpc::ServerContext* context,
 
     auto nopbResponse = std::get<0>(transitionOutcome);
     *response = nopbResponse;
-    OLOG(info) << "Transition response: " << response->state << "ok: " << response->ok;
+    OLOG(info) << "Transition response: " << response->state << " ok: " << response->ok;
 
     return grpc::Status::OK;
 }

Original file line number	Diff line number	Diff line change
`@@ -106,7 +106,7 @@ ::grpc::Status OccLite::Service::Transition(::grpc::ServerContext* context,`
`106`	`106`
`107`	`107`	`auto nopbResponse = std::get<0>(transitionOutcome);`
`108`	`108`	`*response = nopbResponse;`
`109`		`- OLOG(info) << "Transition response: " << response->state << "ok: " << response->ok;`
	`109`	`+ OLOG(info) << "Transition response: " << response->state << " ok: " << response->ok;`
`110`	`110`
`111`	`111`	`return grpc::Status::OK;`
`112`	`112`	`}`