feat(auto): live engine events, forward-compatible decoding, process control

auto/project.scala

@@ -7,14 +7,15 @@
 //> using dep org.virtuslab::besom-core:0.5.2-SNAPSHOT
 //> using dep org.virtuslab::besom-model:0.5.2-SNAPSHOT
 //> using dep org.virtuslab::scala-yaml:0.3.1
+//> using dep com.lihaoyi::geny:1.1.1
 //> using dep com.lihaoyi::os-lib:0.11.8
 //> using dep com.lihaoyi::os-lib-watch:0.11.8
 //> using dep org.eclipse.jgit:org.eclipse.jgit:6.8.0.202311291450-r
 //> using dep org.eclipse.jgit:org.eclipse.jgit.ssh.jsch:6.8.0.202311291450-r
-//> using dep org.slf4j:slf4j-nop:2.0.17 // TODO library should not have bindings for slf4j
-//> using dep ma.chinespirit::tailf:0.1.0
+//> using dep ma.chinespirit::tailf:0.2.0
 
 //> using test.dep org.scalameta::munit:1.2.4
+//> using test.dep org.slf4j:slf4j-nop:2.0.17
 
 //> using publish.name "besom-auto"
 //> using publish.organization "org.virtuslab"

auto/src/main/scala/besom/auto/internal/AutoError.scala

@@ -15,6 +15,46 @@ object AutoError:
   def apply(message: String, cause: Throwable) = new AutoError(Some(message), Some(cause))
   def apply(cause: Throwable)                  = new AutoError(None, Some(cause))
 
+/** Raised when a `pulumi` lifecycle operation (up/preview/refresh/destroy) fails.
+  *
+  * The engine event log is parsed best-effort even on the failure path, so [[diagnostics]] and [[failures]] - the data that explains *why*
+  * the operation failed - stay available. The bulky stdout/stderr dump lives on the [[ShellAutoError]] in [[cause]]; the copies here are
+  * for programmatic access.
+  *
+  * @param operation
+  *   the lifecycle operation that failed, one of `preview`, `up`, `refresh`, `destroy`
+  * @param exitCode
+  *   the exit code of the `pulumi` process
+  * @param stdout
+  *   the standard output of the `pulumi` process
+  * @param stderr
+  *   the standard error of the `pulumi` process
+  * @param resourcePreEvents
+  *   the resource operations the engine started before it gave up
+  * @param resourceOperations
+  *   the resource operations that completed before the engine gave up
+  * @param failures
+  *   the resource operations that failed
+  * @param diagnostics
+  *   the diagnostic messages emitted by the engine and the providers
+  * @param parseErrors
+  *   the event log lines that could not be decoded
+  */
+@SerialVersionUID(1L)
+case class OperationFailedError(
+  message: Option[String],
+  cause: Option[Throwable],
+  operation: String,
+  exitCode: Int,
+  stdout: String,
+  stderr: String,
+  resourcePreEvents: List[ResourcePreEvent],
+  resourceOperations: List[ResOutputsEvent],
+  failures: List[ResOpFailedEvent],
+  diagnostics: List[DiagnosticEvent],
+  parseErrors: List[EventLogParseError]
+) extends BaseAutoError(message, cause)
+
 @SerialVersionUID(1L)
 case class ShellAutoError(
   message: Option[String],

auto/src/main/scala/besom/auto/internal/ChildProcess.scala

@@ -0,0 +1,44 @@
+package besom.auto.internal
+
+/** A handle to a `pulumi` process started by besom-auto, handed to callers that opted in with `ShellOption.OnStart` (or the per-operation
+  * `OnProcessStart` options) so that they can implement cancellation.
+  *
+  * The handle is live only for as long as the operation that produced it runs.
+  */
+final class ChildProcess private[auto] (val underlying: os.SubProcess):
+
+  /** The operating system process id. */
+  def pid: Long = underlying.wrapped.pid()
+
+  /** Whether the process is still running. */
+  def isAlive: Boolean = underlying.isAlive()
+
+  /** Sends `SIGINT`, which is the only signal `pulumi` treats as "cancel gracefully" - it logs `^C received; cancelling` and unwinds the
+    * current step rather than dying mid-operation. Sending it a second time is what `pulumi` itself escalates to immediate termination, so
+    * forwarding a terminal's Ctrl-C straight through gives the usual two stage behaviour.
+    *
+    * The JDK exposes no signal API, so on Unix this shells out to `kill -INT`; on Windows, where there is no equivalent, it falls back to
+    * [[terminate]].
+    *
+    * The signal goes to the process besom-auto spawned, not to its descendants. That is exactly right for `pulumi`, which is always spawned
+    * directly and unwinds its own children - but it does mean a caller that interposes a shell (`sh -c "pulumi ..."`) would break
+    * cancellation, since the shell would receive the signal, ignore it while waiting on its foreground child, and never pass it on.
+    */
+  def interrupt(): Unit =
+    if isWindows then terminate()
+    else
+      // best effort - if the process is already gone kill exits non-zero and there is nothing to cancel
+      os.proc("kill", "-INT", pid.toString).call(check = false)
+      ()
+
+  /** Sends `SIGTERM` and, if the process is still alive after the grace period, `SIGKILL`. Note that `pulumi` does *not* treat `SIGTERM` as
+    * a cancellation - use [[interrupt]] for that.
+    */
+  def terminate(): Unit = underlying.destroy()
+
+  /** Sends `SIGKILL` immediately. The stack is very likely to be left with a pending operation. */
+  def kill(): Unit = underlying.destroy(shutdownGracePeriod = 0)
+
+  private def isWindows: Boolean = System.getProperty("os.name", "").toLowerCase.startsWith("windows")
+
+end ChildProcess

auto/src/main/scala/besom/auto/internal/EventLogs.scala

@@ -0,0 +1,148 @@
+package besom.auto.internal
+
+import scala.util.Try
+import scala.util.control.NonFatal
+
+/** A line of the engine event log that could not be decoded.
+  *
+  * Parse errors are collected instead of failing the operation they belong to - Pulumi is free to add new event types and new values of
+  * existing enums at any time and a successful deployment must not be reported as a failure because of it.
+  *
+  * @param lineNumber
+  *   the 1-based number of the offending line in the event log
+  * @param line
+  *   the raw content of the offending line
+  * @param error
+  *   the decoding error
+  */
+case class EventLogParseError(lineNumber: Int, line: String, error: Exception)
+
+/** The result of decoding an engine event log.
+  *
+  * @param events
+  *   all events that could be decoded, in the order the engine emitted them
+  * @param parseErrors
+  *   all lines that could not be decoded
+  */
+case class ParsedEventLog(events: List[EngineEvent], parseErrors: List[EventLogParseError])
+object ParsedEventLog:
+  val empty: ParsedEventLog = ParsedEventLog(Nil, Nil)
+
+/** Everything related to reading Pulumi's `--event-log` file, both post-hoc and live. */
+private[auto] object EventLogs:
+
+  /** Reads and decodes an engine event log.
+    *
+    * Undecodable lines are collected in [[ParsedEventLog.parseErrors]] and never fail the read; only a failure to read the file itself
+    * produces a `Left`.
+    *
+    * @param path
+    *   the path of the event log
+    * @return
+    *   the decoded log or an error if the file could not be read
+    */
+  def parse(path: os.Path): Either[Exception, ParsedEventLog] =
+    Try(os.read.lines(path)).toEither.left
+      .map(e => AutoError(s"Failed to read event log: $path", e))
+      .map { lines =>
+        val (events, errors) = lines.iterator.zipWithIndex
+          .filter { case (line, _) => line.nonEmpty }
+          .foldLeft((List.empty[EngineEvent], List.empty[EventLogParseError])) { case ((events, errors), (line, idx)) =>
+            EngineEvent.fromJson(line) match
+              case Right(event) => (event :: events, errors)
+              case Left(error)  => (events, EventLogParseError(idx + 1, line, error) :: errors)
+          }
+        ParsedEventLog(events.reverse, errors.reverse)
+      }
+  end parse
+
+  /** Finds the [[SummaryEvent]] emitted at the end of an operation.
+    *
+    * @param events
+    *   the decoded events
+    * @param parseErrors
+    *   the parse errors from the same log, used to explain a missing summary caused by a wire format drift
+    * @return
+    *   the summary event or an error if there was none
+    */
+  def summary(events: List[EngineEvent], parseErrors: List[EventLogParseError] = Nil): Either[Exception, SummaryEvent] =
+    events
+      .collectFirst { case e if e.summaryEvent.isDefined => e.summaryEvent.get }
+      .toRight {
+        val suffix =
+          if parseErrors.isEmpty then ""
+          else s" (${parseErrors.size} event log line(s) failed to parse, the summary event may be among them)"
+        AutoError(s"No summary event found in event log$suffix")
+      }
+  end summary
+
+  def resourcePreEvents(events: List[EngineEvent]): List[ResourcePreEvent] = events.flatMap(_.resourcePreEvent)
+  def resourceOutputs(events: List[EngineEvent]): List[ResOutputsEvent]    = events.flatMap(_.resOutputsEvent)
+  def failures(events: List[EngineEvent]): List[ResOpFailedEvent]          = events.flatMap(_.resOpFailedEvent)
+  def diagnostics(events: List[EngineEvent]): List[DiagnosticEvent]        = events.flatMap(_.diagnosticEvent)
+
+  /** Runs `body` while tailing `path`, handing every decoded engine event to `handler` as it is written.
+    *
+    * The follower is opened before `body` runs, so no event written by the subprocess can be missed. When `body` returns the follower is
+    * stopped, which drains whatever is already on disk before signalling EOF, so events written just before the process exited are still
+    * delivered.
+    *
+    * Undecodable lines are dropped here - [[parse]] accounts for them post-hoc in `parseErrors`. Exceptions thrown by `handler` are
+    * swallowed so that a single misbehaving consumer cannot end the stream.
+    *
+    * Every event is delivered before this returns. `stop()` takes effect at EOF and the reader re-checks it after each `rereadSleep`
+    * (100ms), so the wait is short - but it is unbounded, which means a `handler` that never returns blocks the operation from returning.
+    * That is the reason handlers must not block.
+    *
+    * Caveat: `tailf`'s follower treats a file that got shorter than the current read position as rotated and restarts from offset 0.
+    * Nothing truncates a Pulumi event log mid-run, but it is the only path that could replay events, so consumers should stay idempotent
+    * per URN.
+    *
+    * @param path
+    *   the event log to tail, which must already exist
+    * @param handler
+    *   the consumer of the events, or `None` to run `body` without tailing at all
+    * @param body
+    *   the operation producing the events
+    * @return
+    *   the result of `body`, or a `Left` if the follower could not be opened
+    */
+  def around[A](path: os.Path, handler: Option[EngineEvent => Unit])(body: => Either[Exception, A]): Either[Exception, A] =
+    handler match
+      case None                        => body
+      case Some(_) if !os.exists(path) =>
+        // tailf would happily wait for the file to appear; for us its absence is a setup error, not something to wait out
+        Left(AutoError(s"Cannot stream engine events, event log does not exist: $path"))
+      case Some(onEvent) =>
+        shell.tail(path).flatMap { follower =>
+          val reader = new Thread(
+            new Runnable:
+              def run(): Unit =
+                try
+                  val lines = scala.io.Source.fromInputStream(follower)(using scala.io.Codec.UTF8).getLines()
+                  while lines.hasNext do
+                    val line = lines.next()
+                    if line.nonEmpty then
+                      EngineEvent.fromJson(line) match
+                        case Right(event) =>
+                          try onEvent(event)
+                          catch case NonFatal(_) => () // a broken consumer must not end the stream
+                        case Left(_) => () // accounted for post-hoc by parse
+                catch case NonFatal(_) => () // the follower was closed from under us, nothing left to read
+            ,
+            s"besom-auto-event-tail-${path.last}"
+          )
+          reader.setDaemon(true)
+          reader.start()
+
+          try body
+          finally
+            follower.stop() // takes effect at EOF, so the reader drains what is already written and then ends
+            try reader.join() // no deadline: cutting the reader short here would silently drop events it still holds
+            finally
+              try follower.close()
+              catch case NonFatal(_) => () // releasing the fd must not mask the operation's own outcome
+        }
+  end around
+
+end EventLogs

auto/src/main/scala/besom/auto/internal/Events.scala

@@ -13,6 +13,12 @@ enum DiffKind(val value: String):
   case Update extends DiffKind("update")
   case UpdateReplace extends DiffKind("update-replace")
 
+  /** A diff kind emitted by the engine that this version of besom-auto does not know about.
+    *
+    * Kept so that a newer Pulumi CLI can not fail decoding of an otherwise valid event log.
+    */
+  case Other(unknownValue: String) extends DiffKind(unknownValue)
+
   def forcesReplacement: Boolean = value.endsWith("-replace")
 end DiffKind
 object DiffKind:
@@ -23,7 +29,7 @@ object DiffKind:
     case "delete-replace" => DiffKind.DeleteReplace
     case "update"         => DiffKind.Update
     case "update-replace" => DiffKind.UpdateReplace
-    case other            => throw DeserializationException(s"Unknown DiffKind: $other")
+    case other            => DiffKind.Other(other)
 
   given RootJsonFormat[DiffKind] with
     def write(obj: DiffKind): JsValue = JsString(obj.value)
@@ -36,12 +42,18 @@ end DiffKind
 enum ProgressType(val value: String):
   case PluginDownload extends ProgressType("plugin-download")
   case PluginInstall extends ProgressType("plugin-install")
+
+  /** A progress type emitted by the engine that this version of besom-auto does not know about.
+    *
+    * Kept so that a newer Pulumi CLI can not fail decoding of an otherwise valid event log.
+    */
+  case Other(unknownValue: String) extends ProgressType(unknownValue)
 end ProgressType
 object ProgressType:
   def from(value: String): ProgressType = value match
     case "plugin-download" => ProgressType.PluginDownload
     case "plugin-install"  => ProgressType.PluginInstall
-    case other             => throw DeserializationException(s"Unknown ProgressType: $other")
+    case other             => ProgressType.Other(other)
 
   given RootJsonFormat[ProgressType] with
     def write(obj: ProgressType): JsValue = JsString(obj.value)