feat(api): Standardize mechanism for ResourceHandlers to notify artifact deployments (#1466)

* refactor(api): Move artifact deployment events to keel-api

* feat(api): Add artifact deployment notification methods to ResourceHandler

* fix(pr): Use EventPublisher where needed

Author: luispollo

keel-api/src/main/kotlin/com/netflix/spinnaker/keel/api/events/artifacts.kt

@@ -25,3 +25,19 @@ data class ArtifactRegisteredEvent(
 data class ArtifactSyncEvent(
   val controllerTriggered: Boolean = false
 )
+
+/**
+ * An event fired to signal that an artifact version is deploying to a resource.
+ */
+data class ArtifactVersionDeploying(
+  val resourceId: String,
+  val artifactVersion: String
+)
+
+/**
+ * An event fired to signal that an artifact version was successfully deployed to a resource.
+ */
+data class ArtifactVersionDeployed(
+  val resourceId: String,
+  val artifactVersion: String
+)

keel-api/src/main/kotlin/com/netflix/spinnaker/keel/api/plugins/ResourceHandler.kt

@@ -7,6 +7,11 @@ import com.netflix.spinnaker.keel.api.ResourceKind
 import com.netflix.spinnaker.keel.api.ResourceSpec
 import com.netflix.spinnaker.keel.api.actuation.Task
 import com.netflix.spinnaker.keel.api.artifacts.DeliveryArtifact
+import com.netflix.spinnaker.keel.api.artifacts.PublishedArtifact
+import com.netflix.spinnaker.keel.api.events.ArtifactPublishedEvent
+import com.netflix.spinnaker.keel.api.events.ArtifactVersionDeployed
+import com.netflix.spinnaker.keel.api.events.ArtifactVersionDeploying
+import com.netflix.spinnaker.keel.api.support.EventPublisher
 import com.netflix.spinnaker.kork.exceptions.SystemException
 import com.netflix.spinnaker.kork.plugins.api.internal.SpinnakerExtensionPoint
 
@@ -25,6 +30,11 @@ import com.netflix.spinnaker.kork.plugins.api.internal.SpinnakerExtensionPoint
  * 3. Act to resolve the drift when requested by keel. This is done via the [create], [update] and [delete]
  *    methods, which receive a [ResourceDiff] as a parameter.
  *
+ *    a. A [ResourceHandler] whose [Resource] type supports [DeliveryArtifact] deployments is additionally
+ *       responsible for notifying core Keel when new versions of those artifacts transition to a deploying
+ *       state and to a successfully deployed state via the [notifyArtifactDeploying] and [notifyArtifactDeployed]
+ *       methods, respectively.
+ *
  * @param S the spec type.
  * @param R the resolved model type.
  */
@@ -38,6 +48,12 @@ interface ResourceHandler<S : ResourceSpec, R : Any> : SpinnakerExtensionPoint {
    */
   val supportedKind: SupportedKind<S>
 
+  /**
+   * Optional [EventPublisher] which can be used to signal other modules of relevant events.
+   */
+  val eventPublisher: EventPublisher?
+    get() = null
+
   /**
    * Resolve and convert the resource spec into the type that represents the diff-able desired
    * state.
@@ -137,6 +153,26 @@ interface ResourceHandler<S : ResourceSpec, R : Any> : SpinnakerExtensionPoint {
    */
   @JvmDefault
   suspend fun actuationInProgress(resource: Resource<S>): Boolean = false
+
+  /**
+   * Notifies core Keel that a new artifact version is being deployed to the specified resource.
+   *
+   * The default implementation achieves this by publishing an [ArtifactVersionDeploying] event via the
+   * [EventPublisher], and should *not* be overridden by plugin implementations.
+   */
+  @JvmDefault
+  fun notifyArtifactDeploying(resource: Resource<S>, artifactVersion: String) =
+    eventPublisher?.publishEvent(ArtifactVersionDeploying(resource.id, artifactVersion))
+
+  /**
+   * Notifies core Keel that a new artifact version was successfully deployed to the specified resource.
+   *
+   * The default implementation achieves this by publishing an [ArtifactVersionDeployed] event via the
+   * [EventPublisher], and should *not* be overridden by plugin implementations.
+   */
+  @JvmDefault
+  fun notifyArtifactDeployed(resource: Resource<S>, artifactVersion: String) =
+    eventPublisher?.publishEvent(ArtifactVersionDeployed(resource.id, artifactVersion))
 }
 
 /**

keel-artifact/src/main/kotlin/com/netflix/spinnaker/keel/artifacts/ArtifactDeployedListener.kt

@@ -1,6 +1,6 @@
 package com.netflix.spinnaker.keel.artifacts
 
-import com.netflix.spinnaker.keel.events.ArtifactVersionDeployed
+import com.netflix.spinnaker.keel.api.events.ArtifactVersionDeployed
 import com.netflix.spinnaker.keel.persistence.KeelRepository
 import kotlinx.coroutines.runBlocking
 import org.slf4j.LoggerFactory

keel-artifact/src/main/kotlin/com/netflix/spinnaker/keel/artifacts/ArtifactDeployingListener.kt

@@ -1,6 +1,6 @@
 package com.netflix.spinnaker.keel.artifacts
 
-import com.netflix.spinnaker.keel.events.ArtifactVersionDeploying
+import com.netflix.spinnaker.keel.api.events.ArtifactVersionDeploying
 import com.netflix.spinnaker.keel.persistence.KeelRepository
 import kotlinx.coroutines.runBlocking
 import org.slf4j.LoggerFactory

keel-artifact/src/test/kotlin/com/netflix/spinnaker/keel/artifacts/ArtifactDeployedListenerTests.kt

@@ -1,7 +1,7 @@
 package com.netflix.spinnaker.keel.artifacts
 
 import com.netflix.spinnaker.keel.api.Resource
-import com.netflix.spinnaker.keel.events.ArtifactVersionDeployed
+import com.netflix.spinnaker.keel.api.events.ArtifactVersionDeployed
 import com.netflix.spinnaker.keel.persistence.KeelRepository
 import com.netflix.spinnaker.keel.test.DummyResourceSpec
 import com.netflix.spinnaker.keel.test.deliveryConfig

keel-artifact/src/test/kotlin/com/netflix/spinnaker/keel/artifacts/ArtifactDeployingListenerTests.kt

@@ -1,7 +1,7 @@
 package com.netflix.spinnaker.keel.artifacts
 
 import com.netflix.spinnaker.keel.api.Resource
-import com.netflix.spinnaker.keel.events.ArtifactVersionDeploying
+import com.netflix.spinnaker.keel.api.events.ArtifactVersionDeploying
 import com.netflix.spinnaker.keel.persistence.KeelRepository
 import com.netflix.spinnaker.keel.test.DummyResourceSpec
 import com.netflix.spinnaker.keel.test.deliveryConfig

keel-core/src/main/kotlin/com/netflix/spinnaker/keel/actuation/ResourceActuator.kt

@@ -200,7 +200,7 @@ class ResourceActuator(
           }
         }
       } catch (e: ResourceCurrentlyUnresolvable) {
-        log.warn("Resource check for {} failed (hopefully temporarily) due to {}", id, e.message, e)
+        log.warn("Resource check for {} failed (hopefully temporarily) due to {}", id, e.message)
         publisher.publishEvent(ResourceCheckUnresolvable(resource, e, clock))
       } catch (e: Exception) {
         log.error("Resource check for $id failed", e)

keel-core/src/main/kotlin/com/netflix/spinnaker/keel/events/ArtifactVersionDeployed.kt

@@ -18,6 +18,7 @@ package com.netflix.spinnaker.config
 
 import com.netflix.spinnaker.keel.api.actuation.TaskLauncher
 import com.netflix.spinnaker.keel.api.plugins.Resolver
+import com.netflix.spinnaker.keel.api.support.EventPublisher
 import com.netflix.spinnaker.keel.clouddriver.CloudDriverCache
 import com.netflix.spinnaker.keel.clouddriver.CloudDriverService
 import com.netflix.spinnaker.keel.clouddriver.ImageService
@@ -31,12 +32,11 @@ import com.netflix.spinnaker.keel.ec2.resource.ClusterHandler
 import com.netflix.spinnaker.keel.ec2.resource.SecurityGroupHandler
 import com.netflix.spinnaker.keel.orca.ClusterExportHelper
 import com.netflix.spinnaker.keel.orca.OrcaService
-import java.time.Clock
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.boot.context.properties.EnableConfigurationProperties
-import org.springframework.context.ApplicationEventPublisher
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
+import java.time.Clock
 
 @Configuration
 @EnableConfigurationProperties(CanaryConstraintConfigurationProperties::class)
@@ -50,7 +50,7 @@ class EC2Config {
     taskLauncher: TaskLauncher,
     clock: Clock,
     normalizers: List<Resolver<*>>,
-    publisher: ApplicationEventPublisher,
+    eventPublisher: EventPublisher,
     clusterExportHelper: ClusterExportHelper
   ): ClusterHandler =
     ClusterHandler(
@@ -59,7 +59,7 @@ class EC2Config {
       orcaService,
       taskLauncher,
       clock,
-      publisher,
+      eventPublisher,
       normalizers,
       clusterExportHelper
     )

keel-core/src/main/kotlin/com/netflix/spinnaker/keel/events/ArtifactVersionDeploying.kt

@@ -26,10 +26,10 @@ import com.netflix.spinnaker.keel.ec2.toEc2Api
 import com.netflix.spinnaker.keel.model.Job
 import com.netflix.spinnaker.keel.orca.OrcaService
 import com.netflix.spinnaker.keel.retrofit.isNotFound
-import java.time.Duration
 import kotlinx.coroutines.async
 import kotlinx.coroutines.coroutineScope
 import retrofit2.HttpException
+import java.time.Duration
 
 class ApplicationLoadBalancerHandler(
   private val cloudDriverService: CloudDriverService,

keel-ec2-plugin/src/main/kotlin/com/netflix/spinnaker/config/EC2Config.kt

@@ -28,10 +28,10 @@ import com.netflix.spinnaker.keel.diff.toIndividualDiffs
 import com.netflix.spinnaker.keel.model.Job
 import com.netflix.spinnaker.keel.orca.OrcaService
 import com.netflix.spinnaker.keel.retrofit.isNotFound
-import java.time.Duration
 import kotlinx.coroutines.async
 import kotlinx.coroutines.coroutineScope
 import retrofit2.HttpException
+import java.time.Duration
 
 class ClassicLoadBalancerHandler(
   private val cloudDriverService: CloudDriverService,

keel-ec2-plugin/src/main/kotlin/com/netflix/spinnaker/keel/ec2/resource/ApplicationLoadBalancerHandler.kt

@@ -44,6 +44,7 @@ import com.netflix.spinnaker.keel.api.ec2.byRegion
 import com.netflix.spinnaker.keel.api.ec2.resolve
 import com.netflix.spinnaker.keel.api.plugins.ResolvableResourceHandler
 import com.netflix.spinnaker.keel.api.plugins.Resolver
+import com.netflix.spinnaker.keel.api.support.EventPublisher
 import com.netflix.spinnaker.keel.api.withDefaultsOmitted
 import com.netflix.spinnaker.keel.artifacts.DebianArtifact
 import com.netflix.spinnaker.keel.clouddriver.CloudDriverCache
@@ -64,8 +65,6 @@ import com.netflix.spinnaker.keel.core.serverGroup
 import com.netflix.spinnaker.keel.diff.toIndividualDiffs
 import com.netflix.spinnaker.keel.ec2.MissingAppVersionException
 import com.netflix.spinnaker.keel.ec2.toEc2Api
-import com.netflix.spinnaker.keel.events.ArtifactVersionDeployed
-import com.netflix.spinnaker.keel.events.ArtifactVersionDeploying
 import com.netflix.spinnaker.keel.exceptions.ActiveServerGroupsException
 import com.netflix.spinnaker.keel.exceptions.ExportError
 import com.netflix.spinnaker.keel.orca.ClusterExportHelper
@@ -77,25 +76,24 @@ import com.netflix.spinnaker.keel.orca.waitStage
 import com.netflix.spinnaker.keel.plugin.buildSpecFromDiff
 import com.netflix.spinnaker.keel.retrofit.isNotFound
 import com.netflix.spinnaker.keel.serialization.configuredObjectMapper
-import java.time.Clock
-import java.time.Duration
-import java.time.Instant
-import java.time.ZoneId
-import java.time.format.DateTimeFormatter
 import kotlinx.coroutines.Deferred
 import kotlinx.coroutines.async
 import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.runBlocking
-import org.springframework.context.ApplicationEventPublisher
 import retrofit2.HttpException
+import java.time.Clock
+import java.time.Duration
+import java.time.Instant
+import java.time.ZoneId
+import java.time.format.DateTimeFormatter
 
 class ClusterHandler(
   private val cloudDriverService: CloudDriverService,
   private val cloudDriverCache: CloudDriverCache,
   private val orcaService: OrcaService,
   private val taskLauncher: TaskLauncher,
   private val clock: Clock,
-  private val publisher: ApplicationEventPublisher,
+  override val eventPublisher: EventPublisher,
   resolvers: List<Resolver<*>>,
   private val clusterExportHelper: ClusterExportHelper
 ) : ResolvableResourceHandler<ClusterSpec, Map<String, ServerGroup>>(resolvers) {
@@ -148,12 +146,7 @@ class ClusterHandler(
       )
 
       if (createDiffs.isNotEmpty()) {
-        publisher.publishEvent(
-          ArtifactVersionDeploying(
-            resourceId = resource.id,
-            artifactVersion = version
-          )
-        )
+        notifyArtifactDeploying(resource, version)
       }
 
       return@coroutineScope deferred.map { it.await() }
@@ -311,12 +304,7 @@ class ClusterHandler(
           )
         }
 
-        publisher.publishEvent(
-          ArtifactVersionDeploying(
-            resourceId = resource.id,
-            artifactVersion = version
-          )
-        )
+        notifyArtifactDeploying(resource, version)
 
         val task = deferred.await()
         priorExecutionId = task.id
@@ -887,12 +875,7 @@ class ClusterHandler(
       // // only publish a successfully deployed event if the server group is healthy
       val appVersion = activeServerGroups.first().launchConfiguration.appVersion
       if (appVersion != null) {
-        publisher.publishEvent(
-          ArtifactVersionDeployed(
-            resourceId = resource.id,
-            artifactVersion = appVersion
-          )
-        )
+        notifyArtifactDeployed(resource, appVersion)
       }
     }
 

keel-ec2-plugin/src/main/kotlin/com/netflix/spinnaker/keel/ec2/resource/ClassicLoadBalancerHandler.kt

@@ -26,11 +26,11 @@ import com.netflix.spinnaker.keel.api.ec2.TerminationPolicy
 import com.netflix.spinnaker.keel.api.ec2.VirtualMachineImage
 import com.netflix.spinnaker.keel.api.ec2.resolve
 import com.netflix.spinnaker.keel.api.plugins.Resolver
+import com.netflix.spinnaker.keel.api.support.EventPublisher
 import com.netflix.spinnaker.keel.artifacts.DebianArtifact
 import com.netflix.spinnaker.keel.clouddriver.CloudDriverCache
 import com.netflix.spinnaker.keel.clouddriver.CloudDriverService
 import com.netflix.spinnaker.keel.clouddriver.model.ActiveServerGroup
-import com.netflix.spinnaker.keel.clouddriver.model.Capacity as ClouddriverCapacity
 import com.netflix.spinnaker.keel.clouddriver.model.Network
 import com.netflix.spinnaker.keel.clouddriver.model.SecurityGroupSummary
 import com.netflix.spinnaker.keel.clouddriver.model.Subnet
@@ -48,11 +48,7 @@ import io.mockk.clearAllMocks
 import io.mockk.coEvery
 import io.mockk.every
 import io.mockk.mockk
-import java.time.Clock
-import java.time.Duration
-import java.util.UUID
 import kotlinx.coroutines.runBlocking
-import org.springframework.context.ApplicationEventPublisher
 import strikt.api.expect
 import strikt.api.expectThat
 import strikt.assertions.hasSize
@@ -62,6 +58,10 @@ import strikt.assertions.isEqualTo
 import strikt.assertions.isNotEmpty
 import strikt.assertions.isNotNull
 import strikt.assertions.isNull
+import java.time.Clock
+import java.time.Duration
+import java.util.UUID
+import com.netflix.spinnaker.keel.clouddriver.model.Capacity as ClouddriverCapacity
 
 internal class ClusterExportTests : JUnit5Minutests {
 
@@ -70,7 +70,7 @@ internal class ClusterExportTests : JUnit5Minutests {
   val orcaService = mockk<OrcaService>()
   val normalizers = emptyList<Resolver<ClusterSpec>>()
   val clock = Clock.systemUTC()
-  val publisher: ApplicationEventPublisher = mockk(relaxUnitFun = true)
+  val publisher: EventPublisher = mockk(relaxUnitFun = true)
   val repository = mockk<KeelRepository>()
   val taskLauncher = OrcaTaskLauncher(
     orcaService,

keel-ec2-plugin/src/main/kotlin/com/netflix/spinnaker/keel/ec2/resource/ClusterHandler.kt

@@ -7,6 +7,7 @@ import com.netflix.spinnaker.keel.api.ec2.ApplicationLoadBalancerSpec
 import com.netflix.spinnaker.keel.api.ec2.CLOUD_PROVIDER
 import com.netflix.spinnaker.keel.api.ec2.EC2_APPLICATION_LOAD_BALANCER_V1_1
 import com.netflix.spinnaker.keel.api.plugins.Resolver
+import com.netflix.spinnaker.keel.api.support.EventPublisher
 import com.netflix.spinnaker.keel.clouddriver.CloudDriverCache
 import com.netflix.spinnaker.keel.clouddriver.CloudDriverService
 import com.netflix.spinnaker.keel.clouddriver.model.ApplicationLoadBalancerModel
@@ -40,23 +41,22 @@ import io.mockk.confirmVerified
 import io.mockk.every
 import io.mockk.mockk
 import io.mockk.slot
-import java.util.UUID
 import kotlinx.coroutines.runBlocking
-import org.springframework.context.ApplicationEventPublisher
 import strikt.api.expectThat
 import strikt.assertions.first
 import strikt.assertions.get
 import strikt.assertions.getValue
 import strikt.assertions.isEqualTo
 import strikt.assertions.isFalse
 import strikt.assertions.isTrue
+import java.util.UUID
 
 @Suppress("UNCHECKED_CAST")
 internal class ApplicationLoadBalancerHandlerTests : JUnit5Minutests {
   private val cloudDriverService = mockk<CloudDriverService>()
   private val cloudDriverCache = mockk<CloudDriverCache>()
   private val orcaService = mockk<OrcaService>()
-  private val publisher: ApplicationEventPublisher = mockk(relaxUnitFun = true)
+  private val publisher: EventPublisher = mockk(relaxUnitFun = true)
   private val repository = mockk<KeelRepository> {
     // we're just using this to get notifications
     every { environmentFor(any()) } returns Environment("test")