refactor(titus): Allow serde of operation descriptions (#4001)

clouddriver-core/clouddriver-core.gradle

@@ -8,6 +8,8 @@ dependencies {
   testAnnotationProcessor "org.projectlombok:lombok"
 
   implementation "net.logstash.logback:logstash-logback-encoder"
+  implementation "com.fasterxml.jackson.module:jackson-module-kotlin"
+  implementation "com.fasterxml.jackson.datatype:jackson-datatype-jsr310"
   implementation "com.netflix.eureka:eureka-client"
   implementation "com.netflix.frigga:frigga"
   implementation "com.netflix.spinnaker.fiat:fiat-api:$fiatVersion"

clouddriver-core/src/main/groovy/com/netflix/spinnaker/clouddriver/config/CloudDriverConfig.groovy

@@ -18,17 +18,18 @@ package com.netflix.spinnaker.clouddriver.config
 
 import com.fasterxml.jackson.annotation.JsonInclude
 import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.datatype.jdk8.Jdk8Module
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule
+import com.fasterxml.jackson.module.kotlin.KotlinModule
 import com.netflix.spectator.api.Registry
 import com.netflix.spinnaker.cats.agent.Agent
 import com.netflix.spinnaker.cats.agent.ExecutionInstrumentation
 import com.netflix.spinnaker.cats.agent.NoopExecutionInstrumentation
 import com.netflix.spinnaker.cats.redis.cache.RedisCacheOptions
 import com.netflix.spinnaker.clouddriver.cache.CacheConfig
-import com.netflix.spinnaker.clouddriver.cache.CustomScheduledAgent
 import com.netflix.spinnaker.clouddriver.cache.NoopOnDemandCacheUpdater
 import com.netflix.spinnaker.clouddriver.cache.OnDemandCacheUpdater
 import com.netflix.spinnaker.clouddriver.core.CloudProvider
-
 import com.netflix.spinnaker.clouddriver.core.NoopAtomicOperationConverter
 import com.netflix.spinnaker.clouddriver.core.NoopCloudProvider
 import com.netflix.spinnaker.clouddriver.core.ProjectClustersService
@@ -127,6 +128,7 @@ class CloudDriverConfig {
         jacksonObjectMapperBuilder.serializationInclusion(JsonInclude.Include.NON_NULL)
         jacksonObjectMapperBuilder.failOnEmptyBeans(false)
         jacksonObjectMapperBuilder.failOnUnknownProperties(false)
+        jacksonObjectMapperBuilder.modules(new Jdk8Module(), new JavaTimeModule(), new KotlinModule())
       }
     }
   }

clouddriver-event/src/main/kotlin/com/netflix/spinnaker/clouddriver/event/AbstractSpinnakerEvent.kt

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2019 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.netflix.spinnaker.clouddriver.event
+
+import com.fasterxml.jackson.annotation.JsonIgnore
+import com.netflix.spinnaker.clouddriver.event.exceptions.UninitializedEventException
+
+abstract class AbstractSpinnakerEvent : SpinnakerEvent {
+  /**
+   * Not a lateinit to make Java/Lombok & Jackson compatibility a little easier, although behavior is exactly the same.
+   */
+  @JsonIgnore
+  private var metadata: EventMetadata? = null
+
+  override fun getMetadata(): EventMetadata {
+    return metadata ?: throw UninitializedEventException()
+  }
+
+  override fun setMetadata(eventMetadata: EventMetadata) {
+    metadata = eventMetadata
+  }
+}

clouddriver-event/src/main/kotlin/com/netflix/spinnaker/clouddriver/event/Aggregate.kt

@@ -23,6 +23,8 @@ package com.netflix.spinnaker.clouddriver.event
  * latest event state; any modification to an [Aggregate] event log will increment this value.
  * When an operation is attempted on an [version] which is not head, the event framework will
  * reject the change.
+ *
+ * TODO(rz): Add `currentSequence` to make resuming aggregate processing in-flight easier.
  */
 class Aggregate(
   val type: String,

clouddriver-event/src/main/kotlin/com/netflix/spinnaker/clouddriver/event/CompositeSpinnakerEvent.kt

@@ -0,0 +1,31 @@
+/*
+ * Copyright 2019 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.netflix.spinnaker.clouddriver.event
+
+import com.fasterxml.jackson.annotation.JsonIgnore
+
+/**
+ * Marks a [SpinnakerEvent] as being constructed of multiple [SpinnakerEvent]s.
+ *
+ * This interface is necessary to correctly hydrate [EventMetadata] on [SpinnakerEvent] before persisting.
+ */
+interface CompositeSpinnakerEvent : SpinnakerEvent {
+  /**
+   * Returns a list of the composed [SpinnakerEvent]s.
+   */
+  @JsonIgnore
+  fun getComposedEvents(): List<SpinnakerEvent>
+}

clouddriver-event/src/main/kotlin/com/netflix/spinnaker/clouddriver/event/EventMetadata.kt

@@ -20,13 +20,19 @@ import java.time.Instant
 /**
  * Metadata for a [SpinnakerEvent].
  *
+ * @param id A unique ID for the event (not used beyond tracing, debugging)
+ * @param aggregateType The type of aggregate the event is for
+ * @param aggregateId The id of the aggregate the event is for
  * @param sequence Auto-incrementing number for event ordering
  * @param originatingVersion The aggregate version that originated this event
  * @param timestamp The time at which the event was created
  * @param serviceVersion The version of the service (clouddriver) that created the event
  * @param source Where/what generated the event
  */
 data class EventMetadata(
+  val id: String,
+  val aggregateType: String,
+  val aggregateId: String,
   val sequence: Long,
   val originatingVersion: Long,
   val timestamp: Instant = Instant.now(),

clouddriver-event/src/main/kotlin/com/netflix/spinnaker/clouddriver/event/SpinnakerEvent.kt

@@ -16,25 +16,17 @@
 package com.netflix.spinnaker.clouddriver.event
 
 import com.fasterxml.jackson.annotation.JsonTypeInfo
-import java.util.UUID
 
 /**
- * The base event class for the event sourcing library.
- *
- * @property id A unique ID for the event. This value is for tracing, rather than loading events
- * @property aggregateType The type of aggregate the event is for
- * @property aggregateId The id of the aggregate the event is for
- * @property metadata Associated metadata about the event; not actually part of the "event proper"
+ * The base type for the eventing library. All library-level code is contained within [EventMetadata].
  */
 @JsonTypeInfo(
   use = JsonTypeInfo.Id.NAME,
   include = JsonTypeInfo.As.PROPERTY,
-  property = "spinEventType"
+  property = "eventType"
 )
-abstract class SpinnakerEvent {
-  val id = UUID.randomUUID().toString()
+interface SpinnakerEvent {
+  fun getMetadata(): EventMetadata
 
-  lateinit var aggregateType: String
-  lateinit var aggregateId: String
-  lateinit var metadata: EventMetadata
+  fun setMetadata(eventMetadata: EventMetadata)
 }

clouddriver-event/src/main/kotlin/com/netflix/spinnaker/clouddriver/event/config/MemoryEventRepositoryConfig.kt

@@ -19,7 +19,7 @@ import com.netflix.spectator.api.Registry
 import com.netflix.spinnaker.clouddriver.event.persistence.EventRepository
 import com.netflix.spinnaker.clouddriver.event.persistence.InMemoryEventRepository
 import org.slf4j.LoggerFactory
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean
 import org.springframework.boot.context.properties.ConfigurationProperties
 import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.context.ApplicationEventPublisher
@@ -34,7 +34,7 @@ import javax.validation.constraints.Min
 import kotlin.reflect.KClass
 
 @Configuration
-@ConditionalOnProperty("spinnaker.clouddriver.eventing.memory-repository.enabled", matchIfMissing = true)
+@ConditionalOnMissingBean(EventRepository::class)
 @EnableConfigurationProperties(MemoryEventRepositoryConfigProperties::class)
 open class MemoryEventRepositoryConfig {
 

clouddriver-event/src/main/kotlin/com/netflix/spinnaker/clouddriver/event/exceptions/AggregateChangeRejectedException.kt

@@ -22,6 +22,12 @@ import com.netflix.spinnaker.kork.exceptions.SystemException
  *
  * The process which originated the event must be retryable.
  */
-class AggregateChangeRejectedException(message: String) : SystemException(message) {
+class AggregateChangeRejectedException(
+  aggregateVersion: Long,
+  originatingVersion: Long
+) : SystemException(
+  "Attempting to save new events against an old aggregate version " +
+    "(version: $aggregateVersion, originatingVersion: $originatingVersion)"
+) {
   override fun getRetryable() = true
 }

clouddriver-event/src/main/kotlin/com/netflix/spinnaker/clouddriver/event/exceptions/EventingException.kt

@@ -0,0 +1,21 @@
+/*
+ * Copyright 2019 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.netflix.spinnaker.clouddriver.event.exceptions
+
+/**
+ * Marker
+ */
+interface EventingException

clouddriver-event/src/main/kotlin/com/netflix/spinnaker/clouddriver/event/exceptions/InvalidEventTypeException.kt

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2019 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.netflix.spinnaker.clouddriver.event.exceptions
+
+import com.netflix.spinnaker.kork.exceptions.IntegrationException
+
+/**
+ * Thrown when a [SpinnakerEvent] cannot be created.
+ */
+class InvalidEventTypeException(cause: Throwable) : IntegrationException(cause), EventingException

clouddriver-event/src/main/kotlin/com/netflix/spinnaker/clouddriver/event/exceptions/UninitializedEventException.kt

@@ -0,0 +1,26 @@
+/*
+ * Copyright 2019 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.netflix.spinnaker.clouddriver.event.exceptions
+
+import com.netflix.spinnaker.kork.exceptions.IntegrationException
+
+/**
+ * Thrown when an event's metadata is attempted to be retrieved before it has been initialized
+ * by the library.
+ */
+class UninitializedEventException : IntegrationException(
+  "Cannot access event metadata before initialization"
+), EventingException

clouddriver-event/src/main/kotlin/com/netflix/spinnaker/clouddriver/event/persistence/EventRepository.kt

@@ -17,6 +17,8 @@ package com.netflix.spinnaker.clouddriver.event.persistence
 
 import com.netflix.spinnaker.clouddriver.event.Aggregate
 import com.netflix.spinnaker.clouddriver.event.SpinnakerEvent
+import javax.validation.constraints.Max
+import javax.validation.constraints.Positive
 
 /**
  * The [EventRepository] is responsible for reading and writing immutable event logs from a persistent store.
@@ -49,8 +51,30 @@ interface EventRepository {
   /**
    * List all aggregates for a given type.
    *
-   * @param aggregateType The aggregate collection name. If not provided, all aggregates will be returned.
-   * @return An unordered list of matching aggregates.
+   * @param criteria The criteria to limit the response by
+   * @return A list of matching aggregates
    */
-  fun listAggregates(aggregateType: String?): List<Aggregate>
+  fun listAggregates(criteria: ListAggregatesCriteria): ListAggregatesResult
+
+  /**
+   * @param aggregateType The type of [Aggregate] to return. If unset, all types will be returned.
+   * @param token The page token to paginate from. It will return the first results
+   * @param perPage The number of [Aggregate]s to return in each response
+   */
+  class ListAggregatesCriteria(
+    val aggregateType: String? = null,
+    val token: String? = null,
+
+    @Positive @Max(1000)
+    val perPage: Int = 100
+  )
+
+  /**
+   * @param aggregates The collection of [Aggregate]s returned
+   * @param nextPageToken The next page token
+   */
+  class ListAggregatesResult(
+    val aggregates: List<Aggregate>,
+    val nextPageToken: String? = null
+  )
 }