Add detailed logging to ExchangeSteps and Attempts services (#1565)

world-federation-of-advertisers · Apr 15, 2024 · ba0f72f · ba0f72f
1 parent 0c18429
commit ba0f72f
Show file tree

Hide file tree

Showing 10 changed files with 218 additions and 32 deletions.
diff --git a/...otlin/org/wfanet/measurement/kingdom/deploy/gcloud/spanner/SpannerExchangeStepsService.kt b/...otlin/org/wfanet/measurement/kingdom/deploy/gcloud/spanner/SpannerExchangeStepsService.kt
@@ -18,7 +18,6 @@ import io.grpc.Status
 import java.time.Clock
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.map
-import org.wfanet.measurement.common.grpc.failGrpc
 import org.wfanet.measurement.common.identity.ExternalId
 import org.wfanet.measurement.common.identity.IdGenerator
 import org.wfanet.measurement.common.toProtoTime
@@ -33,6 +32,7 @@ import org.wfanet.measurement.internal.kingdom.ExchangeWorkflow
 import org.wfanet.measurement.internal.kingdom.GetExchangeStepRequest
 import org.wfanet.measurement.internal.kingdom.StreamExchangeStepsRequest
 import org.wfanet.measurement.internal.kingdom.claimReadyExchangeStepResponse
+import org.wfanet.measurement.kingdom.deploy.gcloud.spanner.common.ExchangeStepNotFoundException
 import org.wfanet.measurement.kingdom.deploy.gcloud.spanner.queries.StreamExchangeSteps
 import org.wfanet.measurement.kingdom.deploy.gcloud.spanner.readers.ExchangeStepAttemptReader
 import org.wfanet.measurement.kingdom.deploy.gcloud.spanner.readers.ExchangeStepReader
@@ -48,14 +48,21 @@ class SpannerExchangeStepsService(
 ) : ExchangeStepsCoroutineImplBase() {
 
   override suspend fun getExchangeStep(request: GetExchangeStepRequest): ExchangeStep {
+    val externalRecurringExchangeId = ExternalId(request.externalRecurringExchangeId)
     val exchangeStepResult =
       ExchangeStepReader()
         .readByExternalIds(
           client.singleUse(),
-          ExternalId(request.externalRecurringExchangeId),
+          externalRecurringExchangeId,
           request.date,
           request.stepIndex,
-        ) ?: failGrpc(Status.NOT_FOUND) { "ExchangeStep not found" }
+        )
+        ?: throw ExchangeStepNotFoundException(
+            externalRecurringExchangeId,
+            request.date,
+            request.stepIndex,
+          )
+          .asStatusRuntimeException(Status.Code.NOT_FOUND, "ExchangeStep not found")
 
     return exchangeStepResult.exchangeStep
   }

diff --git a/...n/org/wfanet/measurement/kingdom/deploy/gcloud/spanner/common/KingdomInternalException.kt b/...n/org/wfanet/measurement/kingdom/deploy/gcloud/spanner/common/KingdomInternalException.kt
@@ -614,7 +614,7 @@ class ExchangeStepAttemptNotFoundException(
     get() =
       mapOf(
         "external_recurring_exchange_id" to externalRecurringExchangeId.value.toString(),
-        "date" to date.toString(),
+        "date" to date.toLocalDate().toString(),
         "step_index" to stepIndex.toString(),
         "attempt_number" to attemptNumber.toString(),
       )
@@ -630,7 +630,7 @@ class ExchangeStepNotFoundException(
     get() =
       mapOf(
         "external_recurring_exchange_id" to externalRecurringExchangeId.value.toString(),
-        "date" to date.toString(),
+        "date" to date.toLocalDate().toString(),
         "step_index" to stepIndex.toString(),
       )
 }

diff --git a/src/main/kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha/BUILD.bazel b/src/main/kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha/BUILD.bazel
@@ -204,6 +204,7 @@ kt_jvm_library(
         "//src/main/kotlin/org/wfanet/measurement/api/v2alpha:principal_server_interceptor",
         "//src/main/kotlin/org/wfanet/measurement/api/v2alpha:resource_key",
         "//src/main/kotlin/org/wfanet/measurement/common/identity",
+        "//src/main/kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha:internal_status_conversion",
         "//src/main/proto/wfa/measurement/api/v2alpha:exchange_step_attempt_kt_jvm_proto",
         "//src/main/proto/wfa/measurement/api/v2alpha:exchange_step_attempts_service_kt_jvm_grpc_proto",
         "//src/main/proto/wfa/measurement/api/v2alpha:exchange_step_kt_jvm_proto",
@@ -227,6 +228,7 @@ kt_jvm_library(
         "//src/main/kotlin/org/wfanet/measurement/api/v2alpha:principal_server_interceptor",
         "//src/main/kotlin/org/wfanet/measurement/api/v2alpha:resource_key",
         "//src/main/kotlin/org/wfanet/measurement/common/identity",
+        "//src/main/kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha:internal_status_conversion",
         "//src/main/proto/wfa/measurement/api/v2alpha:data_provider_kt_jvm_proto",
         "//src/main/proto/wfa/measurement/api/v2alpha:exchange_step_kt_jvm_proto",
         "//src/main/proto/wfa/measurement/api/v2alpha:exchange_steps_service_kt_jvm_grpc_proto",

diff --git a/.../kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha/ExchangeStepAttemptsService.kt b/.../kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha/ExchangeStepAttemptsService.kt
@@ -99,12 +99,10 @@ class ExchangeStepAttemptsService(
         internalExchangeStepAttempts.appendLogEntry(internalRequest)
       } catch (e: StatusException) {
         throw when (e.status.code) {
-            Status.Code.NOT_FOUND -> Status.NOT_FOUND
-            Status.Code.DEADLINE_EXCEEDED -> Status.DEADLINE_EXCEEDED
-            else -> Status.UNKNOWN
-          }
-          .withCause(e)
-          .asRuntimeException()
+          Status.Code.NOT_FOUND -> Status.NOT_FOUND
+          Status.Code.DEADLINE_EXCEEDED -> Status.DEADLINE_EXCEEDED
+          else -> Status.UNKNOWN
+        }.toExternalStatusRuntimeException(e)
       }
     return internalResponse.toExchangeStepAttempt()
   }
@@ -135,12 +133,10 @@ class ExchangeStepAttemptsService(
         internalExchangeStepAttempts.finishExchangeStepAttempt(internalRequest)
       } catch (e: StatusException) {
         throw when (e.status.code) {
-            Status.Code.NOT_FOUND -> Status.NOT_FOUND
-            Status.Code.DEADLINE_EXCEEDED -> Status.DEADLINE_EXCEEDED
-            else -> Status.UNKNOWN
-          }
-          .withCause(e)
-          .asRuntimeException()
+          Status.Code.NOT_FOUND -> Status.NOT_FOUND
+          Status.Code.DEADLINE_EXCEEDED -> Status.DEADLINE_EXCEEDED
+          else -> Status.UNKNOWN
+        }.toExternalStatusRuntimeException(e)
       }
     return internalResponse.toExchangeStepAttempt()
   }
@@ -192,11 +188,9 @@ class ExchangeStepAttemptsService(
         )
       } catch (e: StatusException) {
         throw when (e.status.code) {
-            Status.Code.NOT_FOUND -> permissionDeniedStatus()
-            else -> Status.UNKNOWN
-          }
-          .withCause(e)
-          .asRuntimeException()
+          Status.Code.NOT_FOUND -> Status.PERMISSION_DENIED
+          else -> Status.UNKNOWN
+        }.toExternalStatusRuntimeException(e)
       }
     when (authenticatedPrincipal) {
       is DataProviderPrincipal -> {

diff --git a/src/main/kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha/ExchangeStepsService.kt b/src/main/kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha/ExchangeStepsService.kt
@@ -148,12 +148,10 @@ class ExchangeStepsService(
         )
       } catch (e: StatusException) {
         throw when (e.status.code) {
-            Status.Code.NOT_FOUND -> permissionDeniedStatus()
-            Status.Code.DEADLINE_EXCEEDED -> Status.DEADLINE_EXCEEDED
-            else -> Status.UNKNOWN
-          }
-          .withCause(e)
-          .asRuntimeException()
+          Status.Code.NOT_FOUND -> Status.PERMISSION_DENIED
+          Status.Code.DEADLINE_EXCEEDED -> Status.DEADLINE_EXCEEDED
+          else -> Status.UNKNOWN
+        }.toExternalStatusRuntimeException(e)
       }
     when (authenticatedPrincipal) {
       is DataProviderPrincipal -> {

diff --git a/...ain/kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha/InternalStatusConversion.kt b/...ain/kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha/InternalStatusConversion.kt
@@ -25,6 +25,8 @@ import io.grpc.StatusRuntimeException
 import io.grpc.protobuf.StatusProto
 import org.wfanet.measurement.api.v2alpha.AccountKey
 import org.wfanet.measurement.api.v2alpha.CanonicalExchangeKey
+import org.wfanet.measurement.api.v2alpha.CanonicalExchangeStepAttemptKey
+import org.wfanet.measurement.api.v2alpha.CanonicalExchangeStepKey
 import org.wfanet.measurement.api.v2alpha.CanonicalRecurringExchangeKey
 import org.wfanet.measurement.api.v2alpha.CanonicalRequisitionKey
 import org.wfanet.measurement.api.v2alpha.DataProviderCertificateKey
@@ -334,14 +336,35 @@ fun Status.toExternalStatusRuntimeException(
                 )
               )
               .toName()
-          put("recurring_exchange", recurringExchangeName)
+          put("recurringExchange", recurringExchangeName)
           errorMessage = "RecurringExchange $recurringExchangeName not found."
         }
         ErrorCode.EXCHANGE_STEP_NOT_FOUND -> {
-          errorMessage = "ExchangeStep not found."
+          val exchangeStepName =
+            CanonicalExchangeStepKey(
+                externalIdToApiId(
+                  checkNotNull(errorInfo.metadataMap["external_recurring_exchange_id"]).toLong()
+                ),
+                checkNotNull(errorInfo.metadataMap["date"]),
+                checkNotNull(errorInfo.metadataMap["step_index"]),
+              )
+              .toName()
+          put("exchangeStep", exchangeStepName)
+          errorMessage = "ExchangeStep $exchangeStepName not found."
         }
         ErrorCode.EXCHANGE_STEP_ATTEMPT_NOT_FOUND -> {
-          errorMessage = "ExchangeStepAttempt not found."
+          val exchangeStepAttemptName =
+            CanonicalExchangeStepAttemptKey(
+                externalIdToApiId(
+                  checkNotNull(errorInfo.metadataMap["external_recurring_exchange_id"]).toLong()
+                ),
+                checkNotNull(errorInfo.metadataMap["date"]),
+                checkNotNull(errorInfo.metadataMap["step_index"]),
+                checkNotNull(errorInfo.metadataMap["attempt_number"]),
+              )
+              .toName()
+          put("exchangeStepAttempt", exchangeStepAttemptName)
+          errorMessage = "ExchangeStepAttempt $exchangeStepAttemptName not found."
         }
         ErrorCode.EVENT_GROUP_STATE_ILLEGAL -> {
           errorMessage = "EventGroup not found."

diff --git a/src/test/kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha/BUILD.bazel b/src/test/kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha/BUILD.bazel
@@ -155,6 +155,7 @@ kt_jvm_test(
     srcs = ["ExchangeStepAttemptsServiceTest.kt"],
     test_class = "org.wfanet.measurement.kingdom.service.api.v2alpha.ExchangeStepAttemptsServiceTest",
     deps = [
+        "//src/main/kotlin/org/wfanet/measurement/kingdom/deploy/gcloud/spanner/common",
         "//src/main/kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha:exchange_step_attempts_service",
         "//src/main/proto/wfa/measurement/api/v2alpha:exchange_step_attempts_service_kt_jvm_grpc_proto",
         "//src/main/proto/wfa/measurement/api/v2alpha:exchange_step_kt_jvm_proto",
@@ -177,6 +178,7 @@ kt_jvm_test(
     deps = [
         "//src/main/kotlin/org/wfanet/measurement/api/v2alpha:resource_key",
         "//src/main/kotlin/org/wfanet/measurement/api/v2alpha/testing",
+        "//src/main/kotlin/org/wfanet/measurement/kingdom/deploy/gcloud/spanner/common",
         "//src/main/kotlin/org/wfanet/measurement/kingdom/service/api/v2alpha:exchange_steps_service",
         "//src/main/proto/wfa/measurement/api/v2alpha:data_provider_kt_jvm_proto",
         "//src/main/proto/wfa/measurement/api/v2alpha:exchange_step_kt_jvm_proto",

diff --git a/...lin/org/wfanet/measurement/kingdom/service/api/v2alpha/ExchangeStepAttemptsServiceTest.kt b/...lin/org/wfanet/measurement/kingdom/service/api/v2alpha/ExchangeStepAttemptsServiceTest.kt
@@ -31,12 +31,15 @@ import org.junit.runners.JUnit4
 import org.mockito.kotlin.any
 import org.mockito.kotlin.stub
 import org.wfanet.measurement.api.v2alpha.CanonicalExchangeStepAttemptKey
+import org.wfanet.measurement.api.v2alpha.CanonicalExchangeStepKey
+import org.wfanet.measurement.api.v2alpha.CanonicalRecurringExchangeKey
 import org.wfanet.measurement.api.v2alpha.DataProviderKey
 import org.wfanet.measurement.api.v2alpha.DataProviderPrincipal
 import org.wfanet.measurement.api.v2alpha.ExchangeStepAttempt
 import org.wfanet.measurement.api.v2alpha.appendExchangeStepAttemptLogEntryRequest
 import org.wfanet.measurement.api.v2alpha.finishExchangeStepAttemptRequest
 import org.wfanet.measurement.api.v2alpha.withPrincipal
+import org.wfanet.measurement.common.grpc.errorInfo
 import org.wfanet.measurement.common.grpc.testing.GrpcTestServerRule
 import org.wfanet.measurement.common.grpc.testing.mockService
 import org.wfanet.measurement.common.identity.ExternalId
@@ -51,6 +54,9 @@ import org.wfanet.measurement.internal.kingdom.ExchangeStepsGrpcKt.ExchangeSteps
 import org.wfanet.measurement.internal.kingdom.exchangeStep as internalExchangeStep
 import org.wfanet.measurement.internal.kingdom.finishExchangeStepAttemptRequest as internalFinishExchangeStepAttemptRequest
 import org.wfanet.measurement.internal.kingdom.getExchangeStepRequest
+import org.wfanet.measurement.kingdom.deploy.gcloud.spanner.common.ExchangeStepAttemptNotFoundException
+import org.wfanet.measurement.kingdom.deploy.gcloud.spanner.common.ExchangeStepNotFoundException
+import org.wfanet.measurement.kingdom.deploy.gcloud.spanner.common.RecurringExchangeNotFoundException
 
 private const val RECURRING_EXCHANGE_ID = 1L
 private const val STEP_INDEX = 1
@@ -64,6 +70,23 @@ private const val DEBUG_LOG_2_MESSAGE = "some other message"
 
 private val DATE = Date.newBuilder().setYear(2021).setMonth(3).setDay(14).build()
 private const val EXCHANGE_ID = "2021-03-14"
+private val EXTERNAL_DATA_PROVIDER_ID = ExternalId(12345L)
+private val DATA_PROVIDER_KEY = DataProviderKey(EXTERNAL_DATA_PROVIDER_ID.apiId.value)
+private val RECURRING_EXCHANGE_KEY =
+  CanonicalRecurringExchangeKey(ExternalId(RECURRING_EXCHANGE_ID).apiId.value)
+private val EXCHANGE_STEP_KEY =
+  CanonicalExchangeStepKey(
+    recurringExchangeId = RECURRING_EXCHANGE_KEY.recurringExchangeId,
+    exchangeId = EXCHANGE_ID,
+    exchangeStepId = STEP_INDEX.toString(),
+  )
+private val EXCHANGE_STEP_ATTEMPT_KEY =
+  CanonicalExchangeStepAttemptKey(
+    recurringExchangeId = RECURRING_EXCHANGE_KEY.recurringExchangeId,
+    exchangeId = EXCHANGE_ID,
+    exchangeStepId = STEP_INDEX.toString(),
+    exchangeStepAttemptId = ATTEMPT_NUMBER.toString(),
+  )
 
 private val INTERNAL_EXCHANGE_STEP_ATTEMPT: InternalExchangeStepAttempt =
   InternalExchangeStepAttempt.newBuilder()
@@ -285,4 +308,105 @@ class ExchangeStepAttemptsServiceTest {
         }
       )
   }
+
+  @Test
+  fun `finishExchangeStepAttempt throws NOT_FOUND with recurring exchange name when recurring exchange not found`() {
+    internalExchangeSteps.stub {
+      onBlocking { getExchangeStep(any()) }
+        .thenReturn(
+          internalExchangeStep { this.externalDataProviderId = EXTERNAL_DATA_PROVIDER_ID.value }
+        )
+    }
+    internalExchangeStepAttempts.stub {
+      onBlocking { finishExchangeStepAttempt(any()) }
+        .thenThrow(
+          RecurringExchangeNotFoundException(ExternalId(RECURRING_EXCHANGE_ID))
+            .asStatusRuntimeException(Status.Code.NOT_FOUND)
+        )
+    }
+    val exception =
+      assertFailsWith<StatusRuntimeException> {
+        withPrincipal(DataProviderPrincipal(DATA_PROVIDER_KEY)) {
+          runBlocking {
+            service.finishExchangeStepAttempt(
+              finishExchangeStepAttemptRequest {
+                name = EXCHANGE_STEP_ATTEMPT.name
+                finalState = ExchangeStepAttempt.State.FAILED
+                logEntries += EXCHANGE_STEP_ATTEMPT.debugLogEntriesList
+              }
+            )
+          }
+        }
+      }
+    assertThat(exception.status.code).isEqualTo(Status.Code.NOT_FOUND)
+    assertThat(exception.errorInfo?.metadataMap)
+      .containsEntry("recurringExchange", RECURRING_EXCHANGE_KEY.toName())
+  }
+
+  @Test
+  fun `finishExchangeStepAttempt throws PERMISSION_DENIED with exchange step name when exchange step not found`() {
+    internalExchangeSteps.stub {
+      onBlocking { getExchangeStep(any()) }
+        .thenThrow(
+          ExchangeStepNotFoundException(ExternalId(RECURRING_EXCHANGE_ID), DATE, STEP_INDEX)
+            .asStatusRuntimeException(Status.Code.NOT_FOUND)
+        )
+    }
+    val exception =
+      assertFailsWith<StatusRuntimeException> {
+        withPrincipal(DataProviderPrincipal(DATA_PROVIDER_KEY)) {
+          runBlocking {
+            service.finishExchangeStepAttempt(
+              finishExchangeStepAttemptRequest {
+                name = EXCHANGE_STEP_ATTEMPT.name
+                finalState = ExchangeStepAttempt.State.FAILED
+                logEntries += EXCHANGE_STEP_ATTEMPT.debugLogEntriesList
+              }
+            )
+          }
+        }
+      }
+    assertThat(exception.status.code).isEqualTo(Status.Code.PERMISSION_DENIED)
+    assertThat(exception.errorInfo?.metadataMap)
+      .containsEntry("exchangeStep", EXCHANGE_STEP_KEY.toName())
+  }
+
+  @Test
+  fun `finishExchangeStepAttempt throws NOT_FOUND with exchange step attempt name when exchange step attempt not found`() {
+    internalExchangeSteps.stub {
+      onBlocking { getExchangeStep(any()) }
+        .thenReturn(
+          internalExchangeStep { this.externalDataProviderId = EXTERNAL_DATA_PROVIDER_ID.value }
+        )
+    }
+    internalExchangeStepAttempts.stub {
+      onBlocking { finishExchangeStepAttempt(any()) }
+        .thenThrow(
+          ExchangeStepAttemptNotFoundException(
+              ExternalId(RECURRING_EXCHANGE_ID),
+              DATE,
+              STEP_INDEX,
+              ATTEMPT_NUMBER,
+            )
+            .asStatusRuntimeException(Status.Code.NOT_FOUND)
+        )
+    }
+    val exception =
+      assertFailsWith<StatusRuntimeException> {
+        withPrincipal(DataProviderPrincipal(DATA_PROVIDER_KEY)) {
+          runBlocking {
+            service.finishExchangeStepAttempt(
+              finishExchangeStepAttemptRequest {
+                name = EXCHANGE_STEP_ATTEMPT.name
+                finalState = ExchangeStepAttempt.State.FAILED
+                logEntries += EXCHANGE_STEP_ATTEMPT.debugLogEntriesList
+              }
+            )
+          }
+        }
+      }
+    assertThat(exception.status.code).isEqualTo(Status.Code.NOT_FOUND)
+    assertThat(exception.errorInfo?.metadataMap)
+      .containsEntry("exchangeStepAttempt", EXCHANGE_STEP_ATTEMPT_KEY.toName())
+  }
 }