Add type registry and cel env caching.

docs/gke/metrics-deployment.md

@@ -42,7 +42,7 @@ This can be done via the Google Cloud Console under "Features", or using the
 gcloud CLI. For example, assuming a cluster named "kingdom":
 
 ```shell
-gcloud beta container clusters update kingdom --enable-managed-prometheus
+gcloud container clusters update kingdom --enable-managed-prometheus
 ```
 
 ## Service Accounts

src/main/k8s/reporting.cue

@@ -116,6 +116,7 @@ package k8s
 						_encryptionKeyPairConfigFileFlag,
 						"--port=8443",
 						"--health-port=8080",
+						"--event-group-metadata-descriptor-cache-duration=1h",
 			] + _tlsArgs + _internalApiTarget.args + _kingdomApiTarget.args
 
 			spec: template: spec: {

src/main/kotlin/org/wfanet/measurement/integration/common/reporting/BUILD.bazel

@@ -40,6 +40,7 @@ kt_jvm_library(
     deps = [
         "//src/main/kotlin/org/wfanet/measurement/integration/common/reporting/identity:metadata_principal_interceptor",
         "//src/main/kotlin/org/wfanet/measurement/reporting/deploy/common/server:reporting_data_server",
+        "//src/main/kotlin/org/wfanet/measurement/reporting/service/api:cel_env_provider",
         "//src/main/kotlin/org/wfanet/measurement/reporting/service/api/v1alpha:event_groups_service",
         "//src/main/kotlin/org/wfanet/measurement/reporting/service/api/v1alpha:reporting_sets_service",
         "//src/main/kotlin/org/wfanet/measurement/reporting/service/api/v1alpha:reports_service",

src/main/kotlin/org/wfanet/measurement/integration/common/reporting/InProcessLifeOfAReportIntegrationTest.kt

@@ -42,11 +42,11 @@ import org.wfanet.measurement.api.v2alpha.EventGroupsGrpcKt
 import org.wfanet.measurement.api.v2alpha.MeasurementConsumer
 import org.wfanet.measurement.api.v2alpha.MeasurementConsumersGrpcKt
 import org.wfanet.measurement.api.v2alpha.MeasurementsGrpcKt
-import org.wfanet.measurement.api.v2alpha.batchGetEventGroupMetadataDescriptorsResponse
 import org.wfanet.measurement.api.v2alpha.certificate
 import org.wfanet.measurement.api.v2alpha.dataProvider
 import org.wfanet.measurement.api.v2alpha.eventGroup
 import org.wfanet.measurement.api.v2alpha.eventGroupMetadataDescriptor
+import org.wfanet.measurement.api.v2alpha.listEventGroupMetadataDescriptorsResponse
 import org.wfanet.measurement.api.v2alpha.listEventGroupsResponse
 import org.wfanet.measurement.api.v2alpha.measurementConsumer
 import org.wfanet.measurement.common.crypto.SigningKeyHandle
@@ -122,9 +122,9 @@ abstract class InProcessLifeOfAReportIntegrationTest {
   private val publicKingdomEventGroupMetadataDescriptorsMock:
     EventGroupMetadataDescriptorsGrpcKt.EventGroupMetadataDescriptorsCoroutineImplBase =
     mockService {
-      onBlocking { batchGetEventGroupMetadataDescriptors(any()) }
+      onBlocking { listEventGroupMetadataDescriptors(any()) }
         .thenReturn(
-          batchGetEventGroupMetadataDescriptorsResponse {
+          listEventGroupMetadataDescriptorsResponse {
             eventGroupMetadataDescriptors += EVENT_GROUP_METADATA_DESCRIPTOR
           }
         )

src/main/kotlin/org/wfanet/measurement/integration/common/reporting/InProcessReportingServer.kt

@@ -19,7 +19,9 @@ import io.grpc.Channel
 import java.io.File
 import java.security.SecureRandom
 import java.security.cert.X509Certificate
+import java.time.Duration
 import java.util.logging.Logger
+import kotlinx.coroutines.Dispatchers
 import org.junit.rules.TestRule
 import org.junit.runner.Description
 import org.junit.runners.model.Statement
@@ -29,6 +31,7 @@ import org.wfanet.measurement.api.v2alpha.EventGroupMetadataDescriptorsGrpcKt.Ev
 import org.wfanet.measurement.api.v2alpha.EventGroupsGrpcKt.EventGroupsCoroutineStub as PublicKingdomEventGroupsCoroutineStub
 import org.wfanet.measurement.api.v2alpha.MeasurementConsumersGrpcKt.MeasurementConsumersCoroutineStub as PublicKingdomMeasurementConsumersCoroutineStub
 import org.wfanet.measurement.api.v2alpha.MeasurementsGrpcKt.MeasurementsCoroutineStub as PublicKingdomMeasurementsCoroutineStub
+import org.wfanet.measurement.api.withAuthenticationKey
 import org.wfanet.measurement.common.crypto.tink.loadPrivateKey
 import org.wfanet.measurement.common.grpc.testing.GrpcTestServerRule
 import org.wfanet.measurement.common.grpc.withVerboseLogging
@@ -42,6 +45,7 @@ import org.wfanet.measurement.internal.reporting.ReportingSetsGrpcKt.ReportingSe
 import org.wfanet.measurement.internal.reporting.ReportsGrpcKt.ReportsCoroutineStub as InternalReportsCoroutineStub
 import org.wfanet.measurement.reporting.deploy.common.server.ReportingDataServer
 import org.wfanet.measurement.reporting.deploy.common.server.ReportingDataServer.Companion.toList
+import org.wfanet.measurement.reporting.service.api.CelEnvCacheProvider
 import org.wfanet.measurement.reporting.service.api.InMemoryEncryptionKeyPairStore
 import org.wfanet.measurement.reporting.service.api.v1alpha.EventGroupsService
 import org.wfanet.measurement.reporting.service.api.v1alpha.ReportingSetsService
@@ -112,11 +116,20 @@ class InProcessReportingServer(
           )
         )
 
+      val celEnvCacheProvider =
+        CelEnvCacheProvider(
+          publicKingdomEventGroupMetadataDescriptorsClient.withAuthenticationKey(
+            measurementConsumerConfig.apiKey
+          ),
+          Duration.ofSeconds(5),
+          Dispatchers.Default,
+        )
+
       listOf(
           EventGroupsService(
               publicKingdomEventGroupsClient,
-              publicKingdomEventGroupMetadataDescriptorsClient,
-              encryptionKeyPairStore
+              encryptionKeyPairStore,
+              celEnvCacheProvider,
             )
             .withMetadataPrincipalIdentities(measurementConsumerConfig),
           ReportingSetsService(internalReportingSetsClient)

src/main/kotlin/org/wfanet/measurement/reporting/deploy/common/server/BUILD.bazel

@@ -36,6 +36,7 @@ kt_jvm_library(
         "//src/main/kotlin/org/wfanet/measurement/common/api:memoizing_principal_lookup",
         "//src/main/kotlin/org/wfanet/measurement/reporting/deploy/common:encryption_key_pair_map",
         "//src/main/kotlin/org/wfanet/measurement/reporting/deploy/common:kingdom_flags",
+        "//src/main/kotlin/org/wfanet/measurement/reporting/service/api:cel_env_provider",
         "//src/main/kotlin/org/wfanet/measurement/reporting/service/api:encryption_key_pair_store",
         "//src/main/kotlin/org/wfanet/measurement/reporting/service/api/v1alpha:akid_principal_lookup",
         "//src/main/kotlin/org/wfanet/measurement/reporting/service/api/v1alpha:event_groups_service",

src/main/kotlin/org/wfanet/measurement/reporting/deploy/common/server/ReportingApiServerFlags.kt

@@ -14,6 +14,7 @@
 
 package org.wfanet.measurement.reporting.deploy.common.server
 
+import java.time.Duration
 import kotlin.properties.Delegates
 import org.wfanet.measurement.reporting.deploy.common.InternalApiFlags
 import picocli.CommandLine
@@ -30,4 +31,16 @@ class ReportingApiServerFlags {
   )
   var debugVerboseGrpcClientLogging by Delegates.notNull<Boolean>()
     private set
+
+  @CommandLine.Option(
+    names = ["--event-group-metadata-descriptor-cache-duration"],
+    description =
+      [
+        "How long the event group metadata descriptors are cached for before refreshing in format 1d1h1m1s1ms1ns"
+      ],
+    defaultValue = "1h",
+    required = false,
+  )
+  lateinit var eventGroupMetadataDescriptorCacheDuration: Duration
+    private set
 }

src/main/kotlin/org/wfanet/measurement/reporting/deploy/common/server/V1AlphaPublicApiServer.kt

@@ -19,24 +19,29 @@ import io.grpc.Channel
 import io.grpc.ServerServiceDefinition
 import java.io.File
 import java.security.SecureRandom
+import kotlinx.coroutines.Dispatchers
 import org.wfanet.measurement.api.v2alpha.CertificatesGrpcKt.CertificatesCoroutineStub as KingdomCertificatesCoroutineStub
 import org.wfanet.measurement.api.v2alpha.DataProvidersGrpcKt.DataProvidersCoroutineStub as KingdomDataProvidersCoroutineStub
 import org.wfanet.measurement.api.v2alpha.EventGroupMetadataDescriptorsGrpcKt.EventGroupMetadataDescriptorsCoroutineStub as KingdomEventGroupMetadataDescriptorsCoroutineStub
 import org.wfanet.measurement.api.v2alpha.EventGroupsGrpcKt.EventGroupsCoroutineStub as KingdomEventGroupsCoroutineStub
 import org.wfanet.measurement.api.v2alpha.MeasurementConsumersGrpcKt.MeasurementConsumersCoroutineStub as KingdomMeasurementConsumersCoroutineStub
 import org.wfanet.measurement.api.v2alpha.MeasurementsGrpcKt.MeasurementsCoroutineStub as KingdomMeasurementsCoroutineStub
+import org.wfanet.measurement.api.withAuthenticationKey
 import org.wfanet.measurement.common.api.PrincipalLookup
 import org.wfanet.measurement.common.api.memoizing
 import org.wfanet.measurement.common.commandLineMain
 import org.wfanet.measurement.common.crypto.SigningCerts
 import org.wfanet.measurement.common.grpc.CommonServer
 import org.wfanet.measurement.common.grpc.buildMutualTlsChannel
 import org.wfanet.measurement.common.grpc.withVerboseLogging
+import org.wfanet.measurement.common.parseTextProto
+import org.wfanet.measurement.config.reporting.MeasurementConsumerConfigs
 import org.wfanet.measurement.internal.reporting.MeasurementsGrpcKt.MeasurementsCoroutineStub as InternalMeasurementsCoroutineStub
 import org.wfanet.measurement.internal.reporting.ReportingSetsGrpcKt.ReportingSetsCoroutineStub as InternalReportingSetsCoroutineStub
 import org.wfanet.measurement.internal.reporting.ReportsGrpcKt.ReportsCoroutineStub as InternalReportsCoroutineStub
 import org.wfanet.measurement.reporting.deploy.common.EncryptionKeyPairMap
 import org.wfanet.measurement.reporting.deploy.common.KingdomApiFlags
+import org.wfanet.measurement.reporting.service.api.CelEnvCacheProvider
 import org.wfanet.measurement.reporting.service.api.InMemoryEncryptionKeyPairStore
 import org.wfanet.measurement.reporting.service.api.v1alpha.AkidPrincipalLookup
 import org.wfanet.measurement.reporting.service.api.v1alpha.EventGroupsService
@@ -76,7 +81,11 @@ private fun run(
       .withVerboseLogging(reportingApiServerFlags.debugVerboseGrpcClientLogging)
 
   val kingdomChannel: Channel =
-    buildMutualTlsChannel(kingdomApiFlags.target, clientCerts, kingdomApiFlags.target)
+    buildMutualTlsChannel(
+        target = kingdomApiFlags.target,
+        clientCerts = clientCerts,
+        hostName = kingdomApiFlags.certHost
+      )
       .withVerboseLogging(reportingApiServerFlags.debugVerboseGrpcClientLogging)
 
   val principalLookup: PrincipalLookup<ReportingPrincipal, ByteString> =
@@ -86,12 +95,27 @@ private fun run(
       )
       .memoizing()
 
+  val measurementConsumerConfigs =
+    parseTextProto(
+      v1AlphaFlags.measurementConsumerConfigFile,
+      MeasurementConsumerConfigs.getDefaultInstance()
+    )
+
+  val apiKey = measurementConsumerConfigs.configsMap.values.first().apiKey
+  val celEnvCacheProvider =
+    CelEnvCacheProvider(
+      KingdomEventGroupMetadataDescriptorsCoroutineStub(kingdomChannel)
+        .withAuthenticationKey(apiKey),
+      reportingApiServerFlags.eventGroupMetadataDescriptorCacheDuration,
+      Dispatchers.Default,
+    )
+
   val services: List<ServerServiceDefinition> =
     listOf(
       EventGroupsService(
           KingdomEventGroupsCoroutineStub(kingdomChannel),
-          KingdomEventGroupMetadataDescriptorsCoroutineStub(kingdomChannel),
-          InMemoryEncryptionKeyPairStore(encryptionKeyPairMap.keyPairs)
+          InMemoryEncryptionKeyPairStore(encryptionKeyPairMap.keyPairs),
+          celEnvCacheProvider,
         )
         .withPrincipalsFromX509AuthorityKeyIdentifiers(principalLookup),
       ReportingSetsService(InternalReportingSetsCoroutineStub(channel))

src/main/kotlin/org/wfanet/measurement/reporting/service/api/BUILD.bazel

@@ -10,3 +10,18 @@ kt_jvm_library(
         "@wfa_common_jvm//src/main/kotlin/org/wfanet/measurement/common/crypto/tink",
     ],
 )
+
+kt_jvm_library(
+    name = "cel_env_provider",
+    srcs = ["CelEnvProvider.kt"],
+    deps = [
+        "//imports/java/org/projectnessie/cel",
+        "//src/main/proto/wfa/measurement/api/v2alpha:event_group_metadata_descriptor_kt_jvm_proto",
+        "//src/main/proto/wfa/measurement/api/v2alpha:event_group_metadata_descriptors_service_kt_jvm_grpc_proto",
+        "//src/main/proto/wfa/measurement/reporting/v1alpha:event_group_kt_jvm_proto",
+        "@wfa_common_jvm//imports/java/com/google/protobuf",
+        "@wfa_common_jvm//imports/kotlin/kotlinx/coroutines:core",
+        "@wfa_common_jvm//imports/kotlin/kotlinx/coroutines:jdk8",
+        "@wfa_common_jvm//src/main/kotlin/org/wfanet/measurement/common",
+    ],
+)