BFD-3738: Track RDA sequence number latency (#2535)

CMSgov · Feb 3, 2025 · eca4638 · eca4638
1 parent 3232fa9
commit eca4638
Show file tree

Hide file tree

Showing 24 changed files with 243 additions and 5 deletions.
diff --git a/...fd-pipeline/bfd-pipeline-app/src/main/java/gov/cms/bfd/pipeline/app/AppConfiguration.java b/...fd-pipeline/bfd-pipeline-app/src/main/java/gov/cms/bfd/pipeline/app/AppConfiguration.java
@@ -377,6 +377,10 @@ public final class AppConfiguration extends BaseAppConfiguration {
       Set.of(
           "FissClaimRdaSink.change.latency.millis",
           "McsClaimRdaSink.change.latency.millis",
+          "FissClaimRdaSink.lastSeq",
+          "McsClaimRdaSink.lastSeq",
+          "FissClaimRdaSink.maxSeq",
+          "McsClaimRdaSink.maxSeq",
           CcwRifLoadJob.Metrics.DATASET_PROCESSING_ACTIVE_TIMER_NAME,
           CcwRifLoadJob.Metrics.DATASET_PROCESSING_TOTAL_TIMER_NAME,
           CcwRifLoadJob.Metrics.MANIFEST_PROCESSING_ACTIVE_TIMER_NAME,

diff --git a/...d-pipeline/bfd-pipeline-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/RdaSink.java b/...d-pipeline/bfd-pipeline-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/RdaSink.java
@@ -2,6 +2,7 @@
 
 import gov.cms.bfd.model.rda.MessageError;
 import gov.cms.model.dsl.codegen.library.DataTransformer;
+import gov.cms.mpsm.rda.v1.ClaimSequenceNumberRange;
 import jakarta.annotation.Nonnull;
 import java.io.IOException;
 import java.time.Duration;
@@ -191,6 +192,13 @@ Optional<TClaim> transformMessage(String apiVersion, TMessage message)
    */
   int writeClaims(Collection<TClaim> objects) throws ProcessingException;
 
+  /**
+   * Updates the available range of sequence numbers.
+   *
+   * @param sequenceNumberRange Current range of sequence numbers.
+   */
+  void updateSequenceNumberRange(ClaimSequenceNumberRange sequenceNumberRange);
+
   /**
    * Return count of records processed since the most recent call to a write method or this method.
    * Calls to this method collect the current value and resets the counter. The sum of this method's

diff --git a/...eline-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/server/EmptyMessageSource.java b/...eline-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/server/EmptyMessageSource.java
@@ -1,5 +1,6 @@
 package gov.cms.bfd.pipeline.rda.grpc.server;
 
+import gov.cms.mpsm.rda.v1.ClaimSequenceNumberRange;
 import java.util.NoSuchElementException;
 
 /**
@@ -45,6 +46,11 @@ public T next() throws Exception {
     throw new NoSuchElementException();
   }
 
+  @Override
+  public ClaimSequenceNumberRange getSequenceNumberRange() {
+    return ClaimSequenceNumberRange.newBuilder().setLower(0).setUpper(0).build();
+  }
+
   @Override
   public void close() throws Exception {}
 }
diff --git a/...e-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/server/ExceptionMessageSource.java b/...e-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/server/ExceptionMessageSource.java
@@ -1,5 +1,6 @@
 package gov.cms.bfd.pipeline.rda.grpc.server;
 
+import gov.cms.mpsm.rda.v1.ClaimSequenceNumberRange;
 import java.util.function.Supplier;
 
 /**
@@ -62,6 +63,11 @@ public T next() throws Exception {
     return source.next();
   }
 
+  @Override
+  public ClaimSequenceNumberRange getSequenceNumberRange() {
+    return ClaimSequenceNumberRange.newBuilder().setLower(0).setUpper(remainingBeforeThrow).build();
+  }
+
   @Override
   public void close() throws Exception {
     source.close();

diff --git a/...peline-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/server/JsonMessageSource.java b/...peline-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/server/JsonMessageSource.java
@@ -3,6 +3,7 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.io.CharSource;
 import com.google.protobuf.util.JsonFormat;
+import gov.cms.mpsm.rda.v1.ClaimSequenceNumberRange;
 import gov.cms.mpsm.rda.v1.FissClaimChange;
 import gov.cms.mpsm.rda.v1.McsClaimChange;
 import java.io.BufferedReader;
@@ -167,6 +168,12 @@ public T next() throws Exception {
     return answer;
   }
 
+  @Override
+  public ClaimSequenceNumberRange getSequenceNumberRange() {
+    // Can't easily support this since it would require deserializing the entire contents
+    return ClaimSequenceNumberRange.newBuilder().setLower(0).setUpper(0).build();
+  }
+
   @Override
   public void close() throws Exception {
     reader.close();

diff --git a/...d-pipeline-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/server/MessageSource.java b/...d-pipeline-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/server/MessageSource.java
@@ -1,5 +1,7 @@
 package gov.cms.bfd.pipeline.rda.grpc.server;
 
+import gov.cms.mpsm.rda.v1.ClaimSequenceNumberRange;
+
 /**
  * Interface for objects that produce message objects from some source (e.g. a file, an array, a
  * database, etc). Mirrors the Iterator protocol but allows for unwrapped exceptions to be passed
@@ -36,4 +38,11 @@ public interface MessageSource<T> extends AutoCloseable {
    * @throws Exception if there is an issue getting the next claim
    */
   MessageSource<T> skipTo(long startingSequenceNumber) throws Exception;
+
+  /**
+   * Returns the current range of sequence numbers.
+   *
+   * @return sequence number range
+   */
+  ClaimSequenceNumberRange getSequenceNumberRange();
 }
diff --git a/...ne-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/server/RandomFissClaimSource.java b/...ne-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/server/RandomFissClaimSource.java
@@ -2,6 +2,7 @@
 
 import com.google.protobuf.Timestamp;
 import gov.cms.mpsm.rda.v1.ChangeType;
+import gov.cms.mpsm.rda.v1.ClaimSequenceNumberRange;
 import gov.cms.mpsm.rda.v1.FissClaimChange;
 import gov.cms.mpsm.rda.v1.RecordSource;
 import java.time.Clock;
@@ -89,6 +90,11 @@ public FissClaimChange next() {
     return change;
   }
 
+  @Override
+  public ClaimSequenceNumberRange getSequenceNumberRange() {
+    return ClaimSequenceNumberRange.newBuilder().setLower(0).setUpper(maxToSend).build();
+  }
+
   @Override
   public void close() {}
 }
diff --git a/...ine-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/server/RandomMcsClaimSource.java b/...ine-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/server/RandomMcsClaimSource.java
@@ -2,6 +2,7 @@
 
 import com.google.protobuf.Timestamp;
 import gov.cms.mpsm.rda.v1.ChangeType;
+import gov.cms.mpsm.rda.v1.ClaimSequenceNumberRange;
 import gov.cms.mpsm.rda.v1.McsClaimChange;
 import gov.cms.mpsm.rda.v1.RecordSource;
 import java.time.Clock;
@@ -87,6 +88,11 @@ public McsClaimChange next() {
     return change;
   }
 
+  @Override
+  public ClaimSequenceNumberRange getSequenceNumberRange() {
+    return ClaimSequenceNumberRange.newBuilder().setLower(0).setUpper(maxToSend).build();
+  }
+
   @Override
   public void close() {}
 }
diff --git a/.../bfd-pipeline-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/server/RdaService.java b/.../bfd-pipeline-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/server/RdaService.java
@@ -50,6 +50,42 @@ public void getVersion(Empty request, StreamObserver<ApiVersion> responseObserve
     }
   }
 
+  @Override
+  public void getFissClaimsSequenceNumberRange(
+      com.google.protobuf.Empty request,
+      StreamObserver<gov.cms.mpsm.rda.v1.ClaimSequenceNumberRange> responseObserver) {
+    LOGGER.info("start getFissClaimsSequenceNumberRange");
+    try (MessageSource<FissClaimChange> source = messageSourceFactory.createFissMessageSource(0)) {
+      responseObserver.onNext(source.getSequenceNumberRange());
+      responseObserver.onCompleted();
+      LOGGER.info("end getFissClaimsSequenceNumberRange");
+    } catch (Exception ex) {
+      responseObserver.onError(Status.fromThrowable(ex).asException());
+      LOGGER.error(
+          "end getFissClaimsSequenceNumberRange call - call failed with exception: message={}",
+          ex.getMessage(),
+          ex);
+    }
+  }
+
+  @Override
+  public void getMcsClaimsSequenceNumberRange(
+      com.google.protobuf.Empty request,
+      StreamObserver<gov.cms.mpsm.rda.v1.ClaimSequenceNumberRange> responseObserver) {
+    LOGGER.info("start getMcsClaimsSequenceNumberRange");
+    try (MessageSource<McsClaimChange> source = messageSourceFactory.createMcsMessageSource(0)) {
+      responseObserver.onNext(source.getSequenceNumberRange());
+      responseObserver.onCompleted();
+      LOGGER.info("end getMcsClaimsSequenceNumberRange");
+    } catch (Exception ex) {
+      responseObserver.onError(Status.fromThrowable(ex).asException());
+      LOGGER.error(
+          "end getMcsClaimsSequenceNumberRange call - call failed with exception: message={}",
+          ex.getMessage(),
+          ex);
+    }
+  }
+
   @Override
   public void getFissClaims(
       ClaimRequest request, StreamObserver<FissClaimChange> responseObserver) {

diff --git a/...grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/server/S3BucketMessageSourceFactory.java b/...grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/server/S3BucketMessageSourceFactory.java
@@ -3,6 +3,7 @@
 import com.google.common.annotations.VisibleForTesting;
 import gov.cms.bfd.pipeline.rda.grpc.RdaChange;
 import gov.cms.bfd.pipeline.sharedutils.s3.S3DirectoryDao;
+import gov.cms.mpsm.rda.v1.ClaimSequenceNumberRange;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.function.Function;
@@ -186,6 +187,11 @@ public synchronized T next() throws Exception {
       return current.next();
     }
 
+    @Override
+    public ClaimSequenceNumberRange getSequenceNumberRange() {
+      return current.getSequenceNumberRange();
+    }
+
     @Override
     public synchronized void close() throws Exception {
       current.close();

diff --git a/...a-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/sink/concurrent/ConcurrentRdaSink.java b/...a-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/sink/concurrent/ConcurrentRdaSink.java
@@ -5,6 +5,7 @@
 import gov.cms.bfd.pipeline.sharedutils.MultiCloser;
 import gov.cms.bfd.pipeline.sharedutils.SequenceNumberTracker;
 import gov.cms.model.dsl.codegen.library.DataTransformer;
+import gov.cms.mpsm.rda.v1.ClaimSequenceNumberRange;
 import jakarta.annotation.Nonnull;
 import java.io.Closeable;
 import java.io.IOException;
@@ -272,6 +273,11 @@ public Optional<Long> readMaxExistingSequenceNumber() throws ProcessingException
     return sink.readMaxExistingSequenceNumber();
   }
 
+  @Override
+  public void updateSequenceNumberRange(ClaimSequenceNumberRange sequenceNumberRange) {
+    sink.updateSequenceNumberRange(sequenceNumberRange);
+  }
+
   /**
    * This method is not implemented since that would bypass the queue used to schedule writes.
    *

diff --git a/...da-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/sink/direct/AbstractClaimRdaSink.java b/...da-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/sink/direct/AbstractClaimRdaSink.java
@@ -13,6 +13,7 @@
 import gov.cms.bfd.pipeline.sharedutils.PipelineApplicationState;
 import gov.cms.bfd.pipeline.sharedutils.TransactionManager;
 import gov.cms.model.dsl.codegen.library.DataTransformer;
+import gov.cms.mpsm.rda.v1.ClaimSequenceNumberRange;
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.DistributionSummary;
 import io.micrometer.core.instrument.MeterRegistry;
@@ -238,6 +239,11 @@ public int writeClaims(Collection<RdaChange<TClaim>> claims) throws ProcessingEx
     return claims.size();
   }
 
+  @Override
+  public void updateSequenceNumberRange(ClaimSequenceNumberRange sequenceNumberRange) {
+    metrics.setMaxSequenceNumber(sequenceNumberRange.getUpper());
+  }
+
   /**
    * Always returns zero since all claims are written synchronously by writeMessages.
    *
@@ -525,7 +531,7 @@ static class Metrics {
     /** Tracks the number of updates per database transaction. */
     private final DistributionSummary dbBatchSize;
 
-    /** Latest sequnce number from writing a batch. * */
+    /** Latest sequence number from writing a batch. * */
     private final AtomicLong latestSequenceNumber;
 
     /** The value returned by the latestSequenceNumber gauge. * */
@@ -534,6 +540,12 @@ static class Metrics {
     /** The number of insert statements executed. */
     private final DistributionSummary insertCount;
 
+    /** Maximum available sequence number. */
+    private final AtomicLong maxSequenceNumber;
+
+    /** The value returned by the maxSequenceNumber gauge. */
+    private final AtomicLong maxSequenceNumberValue;
+
     /**
      * Initializes all the metrics.
      *
@@ -560,6 +572,9 @@ private Metrics(Class<?> klass, MeterRegistry appMetrics) {
       latestSequenceNumber = GAUGES.getGaugeForName(appMetrics, latestSequenceNumberGaugeName);
       latestSequenceNumberValue = GAUGES.getValueForName(latestSequenceNumberGaugeName);
       insertCount = appMetrics.summary(MetricRegistry.name(base, "insertCount"));
+      String maxSequenceNumberGaugeName = MetricRegistry.name(base, "maxSeq");
+      maxSequenceNumber = GAUGES.getGaugeForName(appMetrics, maxSequenceNumberGaugeName);
+      maxSequenceNumberValue = GAUGES.getValueForName(maxSequenceNumberGaugeName);
     }
 
     /**
@@ -571,5 +586,14 @@ private Metrics(Class<?> klass, MeterRegistry appMetrics) {
     void setLatestSequenceNumber(long value) {
       latestSequenceNumberValue.set(value);
     }
+
+    /**
+     * Sets the {@link #maxSequenceNumber}.
+     *
+     * @param value value to set
+     */
+    void setMaxSequenceNumber(long value) {
+      maxSequenceNumberValue.set(value);
+    }
   }
 }
diff --git a/...ne-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/source/FissClaimStreamCaller.java b/...ne-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/source/FissClaimStreamCaller.java
@@ -1,7 +1,9 @@
 package gov.cms.bfd.pipeline.rda.grpc.source;
 
 import com.google.common.base.Preconditions;
+import com.google.protobuf.Empty;
 import gov.cms.mpsm.rda.v1.ClaimRequest;
+import gov.cms.mpsm.rda.v1.ClaimSequenceNumberRange;
 import gov.cms.mpsm.rda.v1.FissClaimChange;
 import gov.cms.mpsm.rda.v1.RDAServiceGrpc;
 import io.grpc.CallOptions;
@@ -45,4 +47,13 @@ public GrpcResponseStream<FissClaimChange> callService(
         ClientCalls.blockingServerStreamingCall(call, request);
     return new GrpcResponseStream<>(call, apiResults);
   }
+
+  @Override
+  public ClaimSequenceNumberRange callSequenceNumberRangeService(
+      ManagedChannel channel, CallOptions callOptions) {
+    final MethodDescriptor<Empty, ClaimSequenceNumberRange> method =
+        RDAServiceGrpc.getGetFissClaimsSequenceNumberRangeMethod();
+    final ClientCall<Empty, ClaimSequenceNumberRange> call = channel.newCall(method, callOptions);
+    return ClientCalls.blockingUnaryCall(call, Empty.getDefaultInstance());
+  }
 }
diff --git a/...ipeline-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/source/GrpcStreamCaller.java b/...ipeline-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/source/GrpcStreamCaller.java
@@ -5,6 +5,7 @@
 import com.google.protobuf.Empty;
 import gov.cms.bfd.pipeline.rda.grpc.RdaServerJob;
 import gov.cms.mpsm.rda.v1.ApiVersion;
+import gov.cms.mpsm.rda.v1.ClaimSequenceNumberRange;
 import gov.cms.mpsm.rda.v1.RDAServiceGrpc;
 import io.grpc.CallOptions;
 import io.grpc.ClientCall;
@@ -62,6 +63,16 @@ public abstract GrpcResponseStream<TResponse> callService(
       ManagedChannel channel, CallOptions callOptions, long startingSequenceNumber)
       throws Exception;
 
+  /**
+   * Calls the service to get the sequence number range.
+   *
+   * @param channel an already open channel to the service being called
+   * @param callOptions the CallOptions object to use for the API call
+   * @return sequence number range
+   */
+  public abstract ClaimSequenceNumberRange callSequenceNumberRangeService(
+      ManagedChannel channel, CallOptions callOptions);
+
   /**
    * Make a call to the server's {@code getVersion()} service and return the version component. Will
    * retry several times if the call fails. Retries allow the job to handle with a race condition

diff --git a/...ine-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/source/McsClaimStreamCaller.java b/...ine-rda-grpc/src/main/java/gov/cms/bfd/pipeline/rda/grpc/source/McsClaimStreamCaller.java
@@ -1,7 +1,9 @@
 package gov.cms.bfd.pipeline.rda.grpc.source;
 
 import com.google.common.base.Preconditions;
+import com.google.protobuf.Empty;
 import gov.cms.mpsm.rda.v1.ClaimRequest;
+import gov.cms.mpsm.rda.v1.ClaimSequenceNumberRange;
 import gov.cms.mpsm.rda.v1.McsClaimChange;
 import gov.cms.mpsm.rda.v1.RDAServiceGrpc;
 import io.grpc.CallOptions;
@@ -44,4 +46,13 @@ public GrpcResponseStream<McsClaimChange> callService(
         ClientCalls.blockingServerStreamingCall(call, request);
     return new GrpcResponseStream<>(call, apiResults);
   }
+
+  @Override
+  public ClaimSequenceNumberRange callSequenceNumberRangeService(
+      ManagedChannel channel, CallOptions callOptions) {
+    final MethodDescriptor<Empty, ClaimSequenceNumberRange> method =
+        RDAServiceGrpc.getGetMcsClaimsSequenceNumberRangeMethod();
+    final ClientCall<Empty, ClaimSequenceNumberRange> call = channel.newCall(method, callOptions);
+    return ClientCalls.blockingUnaryCall(call, Empty.getDefaultInstance());
+  }
 }