refactor: Use separate queries for each bigquery metrics table

[tristanvuong2021]

refactor: Use separate queries for each bigquery metrics table

It will be easier to add additional tables this way. StreamRequisitions also had to be refactored to improve the performance to meet the rpc deadline.

[wfa-reviewable]

This change is 

[kungfucraig]

Reviewable status: 0 of 16 files reviewed, 1 unresolved discussion (waiting on @tristanvuong2021)

---

src/main/kotlin/org/wfanet/measurement/kingdom/deploy/gcloud/job/OperationalMetricsExport.kt line 279 at r2 (raw file):

  }

  private suspend fun exportRequisitions() {

It seems like there is a lot of copied code from the other function. Can they share some logic? For example, and this is just one example, the code that handles the Measurement Type Case.

[tristanvuong2021]

Reviewable status: 0 of 17 files reviewed, 1 unresolved discussion (waiting on @kungfucraig)

---

src/main/kotlin/org/wfanet/measurement/kingdom/deploy/gcloud/job/OperationalMetricsExport.kt line 279 at r2 (raw file):

Previously, kungfucraig (Craig Wright) wrote…

It seems like there is a lot of copied code from the other function. Can they share some logic? For example, and this is just one example, the code that handles the Measurement Type Case.

The only part I could make shared is the Measurement Type part

[kungfucraig]

Reviewed 12 of 16 files at r1, 3 of 4 files at r2, 2 of 2 files at r3, all commit messages.
Reviewable status: complete! all files reviewed, all discussions resolved (waiting on @tristanvuong2021)

[kungfucraig]

Reviewed 3 of 3 files at r4, all commit messages.
Reviewable status: complete! all files reviewed, all discussions resolved (waiting on @stevenwarejones)

[stevenwarejones]

Reviewed 5 of 16 files at r1, 1 of 4 files at r2, 1 of 2 files at r3, 1 of 3 files at r4, all commit messages.
Reviewable status: complete! all files reviewed, all discussions resolved (waiting on @tristanvuong2021)

[SanjayVas]

Reviewed 4 of 16 files at r1, 1 of 7 files at r5, all commit messages.
Reviewable status: 14 of 20 files reviewed, 1 unresolved discussion (waiting on @kungfucraig, @stevenwarejones, and @tristanvuong2021)

---

src/main/kotlin/org/wfanet/measurement/kingdom/deploy/gcloud/spanner/queries/StreamRequisitions.kt line 24 at r5 (raw file):

import org.wfanet.measurement.kingdom.deploy.gcloud.spanner.readers.RequisitionReader

class StreamRequisitions(request: StreamRequisitionsRequest) :

Move these changes to a separate PR. We'll want them regardless of what happens with metrics export.

[kungfucraig]

Reviewed 7 of 7 files at r5, all commit messages.
Reviewable status: all files reviewed, 1 unresolved discussion (waiting on @stevenwarejones and @tristanvuong2021)

[stevenwarejones]

Reviewed 5 of 16 files at r1, 2 of 3 files at r4, 7 of 7 files at r6, all commit messages.
Reviewable status: all files reviewed, 1 unresolved discussion (waiting on @tristanvuong2021)

[tristanvuong2021]

Reviewable status: 16 of 20 files reviewed, 1 unresolved discussion (waiting on @SanjayVas and @stevenwarejones)

---

src/main/kotlin/org/wfanet/measurement/kingdom/deploy/gcloud/spanner/queries/StreamRequisitions.kt line 24 at r5 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

Move these changes to a separate PR. We'll want them regardless of what happens with metrics export.

Done.

[SanjayVas]

Reviewed 1 of 16 files at r1, 2 of 3 files at r4, 1 of 7 files at r6, 4 of 6 files at r7, all commit messages.
Reviewable status: 18 of 20 files reviewed, 2 unresolved discussions (waiting on @stevenwarejones and @tristanvuong2021)

---

src/main/proto/wfa/measurement/internal/kingdom/measurement.proto line 39 at r7 (raw file):

    // ComputationParticipants.
    COMPUTATION = 1;

These changes to the Kingdom internal API should also be in a separate PR.

[tristanvuong2021]

Reviewable status: 11 of 20 files reviewed, 2 unresolved discussions (waiting on @kungfucraig, @SanjayVas, and @stevenwarejones)

---

src/main/proto/wfa/measurement/internal/kingdom/measurement.proto line 39 at r7 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

These changes to the Kingdom internal API should also be in a separate PR.

Done.

[SanjayVas]

Reviewed 2 of 16 files at r1, 2 of 6 files at r7, 6 of 7 files at r8, all commit messages.
Reviewable status: 19 of 20 files reviewed, 5 unresolved discussions (waiting on @stevenwarejones and @tristanvuong2021)

---

src/main/kotlin/org/wfanet/measurement/kingdom/deploy/gcloud/job/OperationalMetricsExport.kt line 153 at r8 (raw file):

              var latestUpdateTime: Timestamp = Timestamp.getDefaultInstance()

              measurementsClient.streamMeasurements(streamMeasurementsRequest).collect { measurement

All gRPC stub calls must handle StatusException at the call site.

---

src/main/kotlin/org/wfanet/measurement/kingdom/deploy/gcloud/job/OperationalMetricsExport.kt line 172 at r8 (raw file):

                    .toInstant()
                    .minusSeconds(measurement.createTime.toInstant().epochSecond)
                    .epochSecond

This is getting the seconds since epoch for the Instant that happens X seconds before, rather than getting the Duration between the two Instants in seconds. The math may come out the same, but it's not describing what you actually want.

Suggestion:

                  Duration.between(measurement.createTime.toInstant(), measurement.updateTime.toInstant()).toSeconds()

---

src/main/kotlin/org/wfanet/measurement/kingdom/deploy/gcloud/job/OperationalMetricsExport.kt line 429 at r8 (raw file):

      apiVersion: String,
    ): MeasurementType {
      val measurementSpec = signedMessage {

Why do you need to build a SignedMessage then unpack it? Just the following should be sufficient

require(Version.fromString(apiVersion) == Version.V2_ALPHA)

val measurementSpec = MeasurementSpec.parseFrom(serializedMeasurementSpec)
return when (measurementSpec.measurementTypeCase) {
  // ...
}

---

src/main/proto/wfa/measurement/internal/kingdom/measurement.proto line 39 at r7 (raw file):

Previously, tristanvuong2021 (Tristan Vuong) wrote…

Done.

I'm still seeing diffs here in this PR.

[tristanvuong2021]

Reviewable status: 16 of 20 files reviewed, 5 unresolved discussions (waiting on @kungfucraig, @SanjayVas, and @stevenwarejones)

---

src/main/kotlin/org/wfanet/measurement/kingdom/deploy/gcloud/job/OperationalMetricsExport.kt line 153 at r8 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

All gRPC stub calls must handle StatusException at the call site.

Done.

---

src/main/kotlin/org/wfanet/measurement/kingdom/deploy/gcloud/job/OperationalMetricsExport.kt line 172 at r8 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

This is getting the seconds since epoch for the Instant that happens X seconds before, rather than getting the Duration between the two Instants in seconds. The math may come out the same, but it's not describing what you actually want.

Done.

---

src/main/kotlin/org/wfanet/measurement/kingdom/deploy/gcloud/job/OperationalMetricsExport.kt line 429 at r8 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

Why do you need to build a SignedMessage then unpack it? Just the following should be sufficient

require(Version.fromString(apiVersion) == Version.V2_ALPHA)

val measurementSpec = MeasurementSpec.parseFrom(serializedMeasurementSpec)
return when (measurementSpec.measurementTypeCase) {
  // ...
}

Done.

---

src/main/proto/wfa/measurement/internal/kingdom/measurement.proto line 39 at r7 (raw file):

Previously, SanjayVas (Sanjay Vasandani) wrote…

I'm still seeing diffs here in this PR.

Done.

[SanjayVas]

Reviewed 7 of 7 files at r9, all commit messages.
Reviewable status: complete! all files reviewed, all discussions resolved (waiting on @stevenwarejones)

[stevenwarejones]

Reviewed 2 of 6 files at r7, 3 of 7 files at r8, 7 of 7 files at r9, all commit messages.
Reviewable status: complete! all files reviewed, all discussions resolved (waiting on @tristanvuong2021)