Add create-validation time to stream provisioning metric

datastream-common/src/main/java/com/linkedin/datastream/common/DatastreamMetadataConstants.java

@@ -89,6 +89,11 @@ public class DatastreamMetadataConstants {
    */
   public static final String CREATION_MS = "system.creation.ms";
 
+  /**
+   * Time spent running create validation in milliseconds.
+   */
+  public static final String CREATE_VALIDATION_TIME_MS = "system.createValidationTime.ms";
+
   /**
    * Position at which the ingestion should start for the datastream.
    */

datastream-server/src/main/java/com/linkedin/datastream/server/Coordinator.java

@@ -86,6 +86,7 @@
 import com.linkedin.datastream.server.providers.ZookeeperCheckpointProvider;
 import com.linkedin.datastream.server.zk.ZkAdapter;
 
+import static com.linkedin.datastream.common.DatastreamMetadataConstants.CREATE_VALIDATION_TIME_MS;
 import static com.linkedin.datastream.common.DatastreamMetadataConstants.CREATION_MS;
 import static com.linkedin.datastream.common.DatastreamMetadataConstants.SYSTEM_DESTINATION_PREFIX;
 import static com.linkedin.datastream.common.DatastreamMetadataConstants.THROUGHPUT_VIOLATING_TOPICS;
@@ -1424,8 +1425,13 @@ private void recordStreamProvisioningTime(Datastream ds) {
       return;
     }
     try {
-      long streamProvisioningTimeMs = System.currentTimeMillis() - Long.parseLong(creationMsStr);
-      _log.info("Datastream {} transitioned to READY in {} ms", ds.getName(), streamProvisioningTimeMs);
+      long creationTimeMs = System.currentTimeMillis() - Long.parseLong(creationMsStr);
+      long createValidationTimeMs = getCreateValidationTimeMs(ds);
+      // Total stream provisioning time is the creation time plus the create-validation time.
+      long streamProvisioningTimeMs = creationTimeMs + createValidationTimeMs;
+
+      _log.info("Stream provisioning time for Datastream {} - {} ms (creation {} ms + create-validation {} ms)",
+          ds.getName(), streamProvisioningTimeMs, creationTimeMs, createValidationTimeMs);
       if (streamProvisioningTimeMs >= 0) {
         _metrics.updateStreamProvisioningTimeHistogram(streamProvisioningTimeMs);
         // Emit the SLO counters atomically: the total provisioned and exactly one of within/outside SLA.
@@ -1442,6 +1448,25 @@ private void recordStreamProvisioningTime(Datastream ds) {
     }
   }
 
+  /**
+   * Returns the create-validation time in milliseconds that is recorded in the
+   * datastream metadata under {@code system.createValidationTime.ms}, or 0 when the property is
+   * absent, not positive, or not a valid long.
+   */
+  @VisibleForTesting
+  static long getCreateValidationTimeMs(Datastream ds) {
+    String validationMsStr = Objects.requireNonNull(ds.getMetadata()).get(CREATE_VALIDATION_TIME_MS);
+    if (validationMsStr == null) {
+      return 0L;
+    }
+    try {
+      long createValidationTimeMs = Long.parseLong(validationMsStr);
+      return createValidationTimeMs > 0 ? createValidationTimeMs : 0L;
+    } catch (NumberFormatException e) {
+      return 0L;
+    }
+  }
+
   private void hardDeleteDatastream(Datastream ds, List<Datastream> activeStreams) {
     String taskPrefix;
     if (DatastreamUtils.containsTaskPrefix(ds)) {

datastream-server/src/main/java/com/linkedin/datastream/server/CoordinatorConfig.java

@@ -59,7 +59,7 @@ public final class CoordinatorConfig {
   public static final int DEFAULT_MARK_DATASTREMS_STOPPED_TIMEOUT_MS = 60 * 1000;
   public static final int DEFAULT_MARK_DATASTREMS_STOPPED_RETRY_PERIOD_MS = 10 * 1000;
   public static final int DEFAULT_LOG_SIZE_LIMIT_IN_BYTES = 1024 * 1024;
-  public static final long DEFAULT_PROVISIONING_SLA_THRESHOLD_MS = Duration.ofMinutes(8).toMillis();
+  public static final long DEFAULT_PROVISIONING_SLA_THRESHOLD_MS = Duration.ofMinutes(10).toMillis();
 
   private final String _cluster;
   private final String _zkAddress;

datastream-server/src/test/java/com/linkedin/datastream/server/TestCoordinatorConfig.java

@@ -5,6 +5,7 @@
  */
 package com.linkedin.datastream.server;
 
+import java.time.Duration;
 import java.util.Properties;
 
 import org.testng.Assert;
@@ -60,6 +61,19 @@ public void testTaskStopTimeoutAndRetryConfigDefault() {
     Assert.assertEquals(CoordinatorConfig.DEFAULT_TASK_STOP_CHECK_TIMEOUT_MS, config.getTaskStopCheckTimeoutMs());
   }
 
+  @Test
+  public void testProvisioningSlaThresholdConfig() {
+    Properties props = new Properties();
+    CoordinatorConfig config = createCoordinatorConfig(props);
+    Assert.assertEquals(config.getProvisioningSlaThresholdMs(),
+        CoordinatorConfig.DEFAULT_PROVISIONING_SLA_THRESHOLD_MS);
+    Assert.assertEquals(config.getProvisioningSlaThresholdMs(), Duration.ofMinutes(10).toMillis());
+
+    props.put(CoordinatorConfig.CONFIG_PROVISIONING_SLA_THRESHOLD_MS, "1000");
+    CoordinatorConfig overridden = createCoordinatorConfig(props);
+    Assert.assertEquals(overridden.getProvisioningSlaThresholdMs(), 1000L);
+  }
+
   @Test
   public void testForceStopStreamsOnFailureConfig() {
     Properties props = new Properties();

datastream-server/src/test/java/com/linkedin/datastream/server/TestCoordinatorStandalone.java

@@ -0,0 +1,58 @@
+/**
+ *  Copyright 2019 LinkedIn Corporation. All rights reserved.
+ *  Licensed under the BSD 2-Clause License. See the LICENSE file in the project root for license information.
+ *  See the NOTICE file in the project root for additional information regarding copyright ownership.
+ */
+package com.linkedin.datastream.server;
+
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import com.linkedin.data.template.StringMap;
+import com.linkedin.datastream.common.Datastream;
+
+import static com.linkedin.datastream.common.DatastreamMetadataConstants.CREATE_VALIDATION_TIME_MS;
+
+
+/**
+ * Unit tests for {@link Coordinator} that run standalone, without an embedded cluster. The full
+ * integration-style test lives in {@code TestCoordinator} in the datastream-server-restli module.
+ */
+@Test
+public class TestCoordinatorStandalone {
+
+  private static Datastream datastreamWithValidationTime(String createValidationTimeMs) {
+    Datastream ds = new Datastream();
+    StringMap metadata = new StringMap();
+    if (createValidationTimeMs != null) {
+      metadata.put(CREATE_VALIDATION_TIME_MS, createValidationTimeMs);
+    }
+    ds.setMetadata(metadata);
+    return ds;
+  }
+
+  @Test
+  public void testGetCreateValidationTimeMsPresent() {
+    // Present and positive -> included as-is
+    Assert.assertEquals(Coordinator.getCreateValidationTimeMs(datastreamWithValidationTime("1500")), 1500L);
+  }
+
+  @Test
+  public void testGetCreateValidationTimeMsAbsent() {
+    // Property not set -> excluded (0)
+    Assert.assertEquals(Coordinator.getCreateValidationTimeMs(datastreamWithValidationTime(null)), 0L);
+  }
+
+  @Test
+  public void testGetCreateValidationTimeMsNonPositive() {
+    // Zero or negative -> excluded (0)
+    Assert.assertEquals(Coordinator.getCreateValidationTimeMs(datastreamWithValidationTime("0")), 0L);
+    Assert.assertEquals(Coordinator.getCreateValidationTimeMs(datastreamWithValidationTime("-5")), 0L);
+  }
+
+  @Test
+  public void testGetCreateValidationTimeMsMalformed() {
+    // Not a valid long -> excluded (0)
+    Assert.assertEquals(Coordinator.getCreateValidationTimeMs(datastreamWithValidationTime("not-a-number")), 0L);
+  }
+}