Update cross-media-measurement-api for max watch duration.

world-federation-of-advertisers · Sep 7, 2023 · c0786aa · c0786aa
1 parent 1c98b2c
commit c0786aa
Show file tree

Hide file tree

Showing 31 changed files with 122 additions and 102 deletions.
diff --git a/build/repositories.bzl b/build/repositories.bzl
@@ -38,13 +38,11 @@ def wfa_measurement_system_repositories():
         version = "0.10.0",
     )
 
-    # DO_NOT_SUBMIT(world-federation-of-advertisers/cross-media-measurement-api/#163):
-    # switch to a release version before submitting the PR.
     wfa_repo_archive(
         name = "wfa_measurement_proto",
         repo = "cross-media-measurement-api",
-        sha256 = "d8f13212e6f56e3bee74263800a8710a99616bedf96be5347bf2e290f83865cd",
-        version = "0.40.0",
+        sha256 = "ec802fd4f218874047a2e6e6b4be0a4718ab2a98551c0edfaacf5f94f856eb84",
+        version = "0.41.0",
     )
 
     wfa_repo_archive(

diff --git a/src/main/k8s/local/metric_spec_config.textproto b/src/main/k8s/local/metric_spec_config.textproto
@@ -29,7 +29,9 @@ watch_duration_params {
     epsilon: 0.001
     delta: 1.0E-12
   }
-  maximum_watch_duration_per_user: 4000
+  maximum_watch_duration_per_user {
+    seconds: 4000
+  }
 }
 reach_vid_sampling_interval {
   width: 0.01

diff --git a/src/main/kotlin/org/wfanet/measurement/api/v2alpha/tools/BUILD.bazel b/src/main/kotlin/org/wfanet/measurement/api/v2alpha/tools/BUILD.bazel
@@ -33,6 +33,7 @@ kt_jvm_library(
         "//src/main/proto/wfa/measurement/api/v2alpha:date_interval_kt_jvm_proto",
         "//src/main/proto/wfa/measurement/api/v2alpha:measurement_spec_kt_jvm_proto",
         "@wfa_common_jvm//imports/java/picocli",
+        "@wfa_common_jvm//src/main/kotlin/org/wfanet/measurement/common",
     ],
 )
 
@@ -96,7 +97,7 @@ java_binary(
 )
 
 kt_jvm_library(
-    name = "benchmark_library",
+    name = "benchmark",
     srcs = [
         "Benchmark.kt",
     ],
@@ -114,6 +115,7 @@ kt_jvm_library(
         "//src/main/proto/wfa/measurement/api/v2alpha:measurement_kt_jvm_proto",
         "//src/main/proto/wfa/measurement/api/v2alpha:measurements_service_kt_jvm_grpc_proto",
         "//src/main/proto/wfa/measurement/api/v2alpha:model_line_kt_jvm_proto",
+        "@wfa_common_jvm//imports/java/org/jetbrains/annotations",
         "@wfa_common_jvm//imports/java/picocli",
         "@wfa_common_jvm//imports/kotlin/com/google/protobuf/kotlin",
         "@wfa_common_jvm//imports/kotlin/kotlinx/coroutines:core",
@@ -128,7 +130,6 @@ kt_jvm_library(
 
 java_binary(
     name = "Benchmark",
-    main_class = "org.wfanet.measurement.api.v2alpha.tools.BenchmarkKt",
-    tags = ["manual"],
-    runtime_deps = [":benchmark_library"],
+    main_class = "org.wfanet.measurement.api.v2alpha.tools.BenchmarkReport",
+    runtime_deps = [":benchmark"],
 )
diff --git a/src/main/kotlin/org/wfanet/measurement/api/v2alpha/tools/Benchmark.kt b/src/main/kotlin/org/wfanet/measurement/api/v2alpha/tools/Benchmark.kt
@@ -63,6 +63,7 @@ import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.runBlocking
+import org.jetbrains.annotations.VisibleForTesting
 import org.wfanet.measurement.api.v2alpha.DataProvidersGrpcKt.DataProvidersCoroutineStub
 import org.wfanet.measurement.api.v2alpha.EncryptionPublicKey
 import org.wfanet.measurement.api.v2alpha.Measurement
@@ -591,7 +592,7 @@ class Benchmark(
   description = ["Benchmark report from Kingdom"],
   sortOptions = false,
 )
-class BenchmarkReport(val clock: Clock = Clock.systemUTC()) : Runnable {
+class BenchmarkReport private constructor(val clock: Clock = Clock.systemUTC()) : Runnable {
   @CommandLine.Mixin private lateinit var tlsFlags: TlsFlags
   @CommandLine.Mixin private lateinit var apiFlags: ApiFlags
   @CommandLine.Mixin private lateinit var baseFlags: BaseFlags
@@ -620,11 +621,17 @@ class BenchmarkReport(val clock: Clock = Clock.systemUTC()) : Runnable {
       Benchmark(baseFlags, createMeasurementFlags, channel, apiAuthenticationKey, clock)
     benchmark.generateBenchmarkReport()
   }
-}
 
-/**
- * Create a benchmarking report of the public Kingdom API.
- *
- * Use the `help` command to see usage details.
- */
-fun main(args: Array<String>) = commandLineMain(BenchmarkReport(), args)
+  companion object {
+    /**
+     * Create a benchmarking report of the public Kingdom API.
+     *
+     * Use the `help` command to see usage details.
+     */
+    fun main(args: Array<String>) = commandLineMain(BenchmarkReport(), args)
+
+    @VisibleForTesting
+    internal fun main(args: Array<String>, clock: Clock) =
+      commandLineMain(BenchmarkReport(clock), args)
+  }
+}
diff --git a/src/main/kotlin/org/wfanet/measurement/api/v2alpha/tools/CreateMeasurementFlags.kt b/src/main/kotlin/org/wfanet/measurement/api/v2alpha/tools/CreateMeasurementFlags.kt
@@ -15,11 +15,13 @@
 package org.wfanet.measurement.api.v2alpha.tools
 
 import java.io.File
+import java.time.Duration
 import java.time.Instant
 import kotlin.properties.Delegates
 import org.wfanet.measurement.api.v2alpha.MeasurementSpec
 import org.wfanet.measurement.api.v2alpha.MeasurementSpecKt
 import org.wfanet.measurement.api.v2alpha.differentialPrivacyParams
+import org.wfanet.measurement.common.toProtoDuration
 import picocli.CommandLine.ArgGroup
 import picocli.CommandLine.Option
 
@@ -266,12 +268,13 @@ class CreateMeasurementFlags {
           var privacyDelta by Delegates.notNull<Double>()
             private set
 
-          @set:Option(
+          @Option(
             names = ["--max-duration"],
-            description = ["Maximum watch duration per user"],
+            description =
+              ["Maximum watch duration per user as a human-readable string, e.g. 5m20s"],
             required = true,
           )
-          var maximumWatchDurationPerUser by Delegates.notNull<Int>()
+          lateinit var maximumWatchDurationPerUser: Duration
             private set
         }
 
@@ -392,6 +395,7 @@ class CreateMeasurementFlags {
       maximumWatchDurationPerUser =
         measurementParams.eventMeasurementParams.eventMeasurementTypeParams.duration
           .maximumWatchDurationPerUser
+          .toProtoDuration()
     }
   }
 

diff --git a/src/main/kotlin/org/wfanet/measurement/api/v2alpha/tools/MeasurementSystem.kt b/src/main/kotlin/org/wfanet/measurement/api/v2alpha/tools/MeasurementSystem.kt
@@ -28,7 +28,7 @@ import java.security.SignatureException
 import java.security.cert.CertPathValidatorException
 import java.security.cert.X509Certificate
 import java.time.Clock
-import java.time.Duration as systemDuration
+import java.time.Duration as JavaDuration
 import java.time.Instant
 import java.time.LocalDate
 import kotlinx.coroutines.CoroutineDispatcher
@@ -157,7 +157,7 @@ import picocli.CommandLine.Parameters
 import picocli.CommandLine.ParentCommand
 import picocli.CommandLine.Spec
 
-private val CHANNEL_SHUTDOWN_TIMEOUT = systemDuration.ofSeconds(30)
+private val CHANNEL_SHUTDOWN_TIMEOUT = JavaDuration.ofSeconds(30)
 
 @Command(
   name = "MeasurementSystem",

diff --git a/...kotlin/org/wfanet/measurement/integration/common/reporting/v2/InProcessReportingServer.kt b/...kotlin/org/wfanet/measurement/integration/common/reporting/v2/InProcessReportingServer.kt
@@ -17,6 +17,7 @@
 package org.wfanet.measurement.integration.common.reporting.v2
 
 import com.google.protobuf.ByteString
+import com.google.protobuf.util.Durations
 import io.grpc.Channel
 import io.grpc.Status
 import io.grpc.StatusException
@@ -294,7 +295,7 @@ class InProcessReportingServer(
               epsilon = 0.001
               delta = 1e-12
             }
-          maximumWatchDurationPerUser = 4000
+          maximumWatchDurationPerUser = Durations.fromSeconds(4000)
         }
       watchDurationVidSamplingInterval =
         MetricSpecConfigKt.vidSamplingInterval {

diff --git a/src/main/kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha/MeasurementsService.kt b/src/main/kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha/MeasurementsService.kt
@@ -316,7 +316,7 @@ private fun MeasurementSpec.validate() {
         "Duration privacy params are invalid"
       }
 
-      grpcRequire(duration.maximumWatchDurationPerUser > 0) {
+      grpcRequire(duration.hasMaximumWatchDurationPerUser()) {
         "Maximum watch duration per user is unspecified"
       }
     }

diff --git a/...otlin/org/wfanet/measurement/loadtest/measurementconsumer/MeasurementConsumerSimulator.kt b/...otlin/org/wfanet/measurement/loadtest/measurementconsumer/MeasurementConsumerSimulator.kt
@@ -17,6 +17,7 @@ package org.wfanet.measurement.loadtest.measurementconsumer
 import com.google.common.truth.Truth.assertThat
 import com.google.protobuf.ByteString
 import com.google.protobuf.Message
+import com.google.protobuf.util.Durations
 import com.google.type.interval
 import io.grpc.StatusException
 import java.security.SignatureException
@@ -612,7 +613,7 @@ class MeasurementConsumerSimulator(
       measurementPublicKey = serializedMeasurementPublicKey
       duration = duration {
         privacyParams = outputDpParams
-        maximumWatchDurationPerUser = 1
+        maximumWatchDurationPerUser = Durations.fromMinutes(1)
       }
       this.nonceHashes += nonceHashes
     }

diff --git a/src/main/kotlin/org/wfanet/measurement/reporting/deploy/v2/postgres/readers/BUILD.bazel b/src/main/kotlin/org/wfanet/measurement/reporting/deploy/v2/postgres/readers/BUILD.bazel
@@ -14,6 +14,7 @@ kt_jvm_library(
         "//src/main/proto/wfa/measurement/internal/reporting/v2:reporting_sets_service_kt_jvm_grpc_proto",
         "//src/main/proto/wfa/measurement/internal/reporting/v2:reports_service_kt_jvm_grpc_proto",
         "@wfa_common_jvm//imports/java/io/r2dbc",
+        "@wfa_common_jvm//imports/java/org/postgresql:r2dbc",
         "@wfa_common_jvm//imports/kotlin/kotlinx/coroutines:core",
         "@wfa_common_jvm//src/main/kotlin/org/wfanet/measurement/common",
         "@wfa_common_jvm//src/main/kotlin/org/wfanet/measurement/common/db/r2dbc",

diff --git a/src/main/kotlin/org/wfanet/measurement/reporting/deploy/v2/postgres/readers/MetricReader.kt b/src/main/kotlin/org/wfanet/measurement/reporting/deploy/v2/postgres/readers/MetricReader.kt
@@ -19,6 +19,7 @@ package org.wfanet.measurement.reporting.deploy.v2.postgres.readers
 import com.google.protobuf.Timestamp
 import com.google.type.Interval
 import com.google.type.interval
+import io.r2dbc.postgresql.codec.Interval as PostgresInterval
 import java.time.Instant
 import java.util.UUID
 import kotlinx.coroutines.flow.Flow
@@ -29,6 +30,7 @@ import org.wfanet.measurement.common.db.r2dbc.ReadContext
 import org.wfanet.measurement.common.db.r2dbc.ResultRow
 import org.wfanet.measurement.common.db.r2dbc.boundStatement
 import org.wfanet.measurement.common.identity.InternalId
+import org.wfanet.measurement.common.toProtoDuration
 import org.wfanet.measurement.common.toProtoTime
 import org.wfanet.measurement.internal.reporting.v2.BatchGetMetricsRequest
 import org.wfanet.measurement.internal.reporting.v2.Measurement
@@ -373,7 +375,7 @@ class MetricReader(private val readContext: ReadContext) {
       val frequencyDifferentialPrivacyDelta: Double? = row["FrequencyDifferentialPrivacyDelta"]
       val maximumFrequency: Int? = row["MaximumFrequency"]
       val maximumFrequencyPerUser: Int? = row["MaximumFrequencyPerUser"]
-      val maximumWatchDurationPerUser: Int? = row["MaximumWatchDurationPerUser"]
+      val maximumWatchDurationPerUser: PostgresInterval? = row["MaximumWatchDurationPerUser"]
       val vidSamplingStart: Float = row["VidSamplingIntervalStart"]
       val vidSamplingWidth: Float = row["VidSamplingIntervalWidth"]
       val createTime: Instant = row["CreateTime"]
@@ -457,18 +459,15 @@ class MetricReader(private val readContext: ReadContext) {
                   }
               }
               MetricSpec.TypeCase.WATCH_DURATION -> {
-                if (maximumWatchDurationPerUser == null) {
-                  throw IllegalStateException()
-                }
-
                 watchDuration =
                   MetricSpecKt.watchDurationParams {
                     privacyParams =
                       MetricSpecKt.differentialPrivacyParams {
                         epsilon = differentialPrivacyEpsilon
                         delta = differentialPrivacyDelta
                       }
-                    this.maximumWatchDurationPerUser = maximumWatchDurationPerUser
+                    this.maximumWatchDurationPerUser =
+                      checkNotNull(maximumWatchDurationPerUser).duration.toProtoDuration()
                   }
               }
               MetricSpec.TypeCase.TYPE_NOT_SET -> throw IllegalStateException()

diff --git a/src/main/kotlin/org/wfanet/measurement/reporting/deploy/v2/postgres/writers/CreateMetrics.kt b/src/main/kotlin/org/wfanet/measurement/reporting/deploy/v2/postgres/writers/CreateMetrics.kt
@@ -16,6 +16,7 @@
 
 package org.wfanet.measurement.reporting.deploy.v2.postgres.writers
 
+import io.r2dbc.postgresql.codec.Interval as PostgresInterval
 import java.time.Instant
 import java.time.ZoneOffset
 import java.util.UUID
@@ -24,6 +25,7 @@ import org.wfanet.measurement.common.db.r2dbc.BoundStatement
 import org.wfanet.measurement.common.db.r2dbc.boundStatement
 import org.wfanet.measurement.common.db.r2dbc.postgres.PostgresWriter
 import org.wfanet.measurement.common.identity.InternalId
+import org.wfanet.measurement.common.toDuration
 import org.wfanet.measurement.common.toInstant
 import org.wfanet.measurement.common.toJson
 import org.wfanet.measurement.common.toProtoTime
@@ -209,7 +211,7 @@ class CreateMetrics(private val requests: List<CreateMetricRequest>) :
                   bind("$11", frequencyHistogram.frequencyPrivacyParams.epsilon)
                   bind("$12", frequencyHistogram.reachPrivacyParams.delta)
                   bind<Long?>("$13", null)
-                  bind<Long?>("$14", null)
+                  bind<PostgresInterval?>("$14", null)
                   bind("$20", frequencyHistogram.maximumFrequency)
                 }
                 MetricSpec.TypeCase.REACH -> {
@@ -219,7 +221,7 @@ class CreateMetrics(private val requests: List<CreateMetricRequest>) :
                   bind<Double?>("$11", null)
                   bind<Double?>("$12", null)
                   bind<Long?>("$13", null)
-                  bind<Long?>("$14", null)
+                  bind<PostgresInterval?>("$14", null)
                   bind<Long?>("$20", null)
                 }
                 MetricSpec.TypeCase.IMPRESSION_COUNT -> {
@@ -229,7 +231,7 @@ class CreateMetrics(private val requests: List<CreateMetricRequest>) :
                   bind<Double?>("$11", null)
                   bind<Double?>("$12", null)
                   bind("$13", impressionCount.maximumFrequencyPerUser)
-                  bind<Long?>("$14", null)
+                  bind<PostgresInterval?>("$14", null)
                   bind<Long?>("$20", null)
                 }
                 MetricSpec.TypeCase.WATCH_DURATION -> {
@@ -239,7 +241,10 @@ class CreateMetrics(private val requests: List<CreateMetricRequest>) :
                   bind<Double?>("$11", null)
                   bind<Double?>("$12", null)
                   bind<Long?>("$13", null)
-                  bind("$14", watchDuration.maximumWatchDurationPerUser)
+                  bind(
+                    "$14",
+                    PostgresInterval.of(watchDuration.maximumWatchDurationPerUser.toDuration())
+                  )
                   bind<Long?>("$20", null)
                 }
                 MetricSpec.TypeCase.TYPE_NOT_SET -> {}

diff --git a/src/main/kotlin/org/wfanet/measurement/reporting/service/api/v1alpha/ReportsService.kt b/src/main/kotlin/org/wfanet/measurement/reporting/service/api/v1alpha/ReportsService.kt
@@ -17,8 +17,7 @@
 package org.wfanet.measurement.reporting.service.api.v1alpha
 
 import com.google.protobuf.ByteString
-import com.google.protobuf.Duration
-import com.google.protobuf.duration
+import com.google.protobuf.Duration as ProtoDuration
 import com.google.protobuf.util.Durations
 import com.google.protobuf.util.Timestamps
 import com.google.type.Interval
@@ -1256,7 +1255,9 @@ class ReportsService(
         InternalMetricTypeCase.WATCH_DURATION -> {
           duration =
             buildDurationMeasurementSpec(
-              internalMetricDetails.watchDuration.maximumWatchDurationPerUser
+              Durations.fromSeconds(
+                internalMetricDetails.watchDuration.maximumWatchDurationPerUserSeconds.toLong()
+              )
             )
           vidSamplingInterval = buildDurationVidSamplingInterval(secureRandom)
         }
@@ -1695,7 +1696,7 @@ private fun buildImpressionMeasurementSpec(
 
 /** Builds a [MeasurementSpec.ReachAndFrequency] for watch duration. */
 private fun buildDurationMeasurementSpec(
-  maximumWatchDurationPerUser: Int
+  maximumWatchDurationPerUser: ProtoDuration
 ): MeasurementSpec.Duration {
   return MeasurementSpecKt.duration {
     privacyParams = differentialPrivacyParams {
@@ -1721,7 +1722,7 @@ private fun SetOperation.Type.toInternal(): InternalSetOperation.Type {
 private fun WatchDurationParams.toInternal(): InternalWatchDurationParams {
   val source = this
   return InternalMetricKt.watchDurationParams {
-    maximumWatchDurationPerUser = source.maximumWatchDurationPerUser
+    maximumWatchDurationPerUserSeconds = source.maximumWatchDurationPerUser
   }
 }
 
@@ -1795,7 +1796,7 @@ private fun buildRowHeader(timeInterval: TimeInterval): String {
   return "$startTimeInstant-$endTimeInstant"
 }
 
-private operator fun Duration.plus(other: Duration): Duration {
+private operator fun ProtoDuration.plus(other: ProtoDuration): ProtoDuration {
   return Durations.add(this, other)
 }
 
@@ -1832,10 +1833,7 @@ private fun aggregateResults(
   var reachValue = 0L
   var impressionValue = 0L
   val frequencyDistribution = mutableMapOf<Long, Double>()
-  var watchDurationValue = duration {
-    seconds = 0
-    nanos = 0
-  }
+  var watchDurationValue = ProtoDuration.getDefaultInstance()
 
   // Aggregation
   for (result in internalResultsList) {
@@ -2078,7 +2076,9 @@ private fun InternalSetOperation.Type.toType(): SetOperation.Type {
 /** Converts an internal [InternalWatchDurationParams] to a public [WatchDurationParams]. */
 private fun InternalWatchDurationParams.toWatchDuration(): WatchDurationParams {
   val source = this
-  return watchDurationParams { maximumWatchDurationPerUser = source.maximumWatchDurationPerUser }
+  return watchDurationParams {
+    maximumWatchDurationPerUser = source.maximumWatchDurationPerUserSeconds
+  }
 }
 
 /** Converts an internal [InternalImpressionCountParams] to a public [ImpressionCountParams]. */

diff --git a/...tlin/org/wfanet/measurement/reporting/service/internal/testing/MeasurementsServiceTest.kt b/...tlin/org/wfanet/measurement/reporting/service/internal/testing/MeasurementsServiceTest.kt
@@ -442,7 +442,7 @@ abstract class MeasurementsServiceTest<T : MeasurementsCoroutineImplBase> {
   fun `setMeasurementResult succeeds in setting the result for report with duration metric`() {
     val metricDetails =
       MetricKt.details {
-        watchDuration = MetricKt.watchDurationParams { maximumWatchDurationPerUser = 100 }
+        watchDuration = MetricKt.watchDurationParams { maximumWatchDurationPerUserSeconds = 100 }
       }
     val createdReport = runBlocking {
       reportsService.createReport(