ui: add line graph for replication lag metric

There currently does not exist observability for replication lag in PCR
in the DB Console. As replication lag is essentially RPO for customers,
this metric should be made available to them in the dashboard. This
commit adds the metric as the difference between the wall time and the
reported replicated time.

Fixes #120652

Release note (ui change): Added observability for PCR replication lag
to the metrics dashboard

pkg/ui/workspaces/db-console/src/views/cluster/containers/nodeGraphs/dashboards/crossClusterReplication.tsx

@@ -12,9 +12,10 @@ import React from "react";
 
 import LineGraph from "src/views/cluster/components/linegraph";
 import { Metric, Axis } from "src/views/shared/components/metricQuery";
-import { AxisUnits } from "@cockroachlabs/cluster-ui";
-
+import { AxisUnits, util } from "@cockroachlabs/cluster-ui";
 import { GraphDashboardProps } from "./dashboardUtils";
+import { cockroach } from "src/js/protos";
+import TimeSeriesQueryAggregator = cockroach.ts.tspb.TimeSeriesQueryAggregator;
 
 export default function (props: GraphDashboardProps) {
   const { storeSources, tenantSource } = props;
@@ -48,5 +49,34 @@ export default function (props: GraphDashboardProps) {
         />
       </Axis>
     </LineGraph>,
+    <LineGraph
+      title="Replication Lag"
+      sources={storeSources}
+      tenantSource={tenantSource}
+      tooltip={`Replication lag between primary and standby cluster`}
+    >
+      <Axis units={AxisUnits.Duration} label="duration">
+        <Metric
+          downsampler={TimeSeriesQueryAggregator.MIN}
+          aggregator={TimeSeriesQueryAggregator.AVG}
+          name="cr.node.physical_replication.replicated_time_seconds"
+          title="Replication Lag"
+          transform={datapoints =>
+            datapoints
+              .filter(d => d.value !== 0)
+              .map(d =>
+                d.value
+                  ? {
+                      ...d,
+                      value:
+                        d.timestamp_nanos.toNumber() -
+                        util.SecondsToNano(d.value),
+                    }
+                  : d,
+              )
+          }
+        />
+      </Axis>
+    </LineGraph>,
   ];
 }

pkg/ui/workspaces/db-console/src/views/cluster/util/graphs.ts

@@ -64,14 +64,18 @@ export function formatMetricData(
   _.each(metrics, (s, idx) => {
     const result = data.results[idx];
     if (result && canShowMetric(result)) {
-      const scaledValues = result.datapoints.map(v => ({
-        ...v,
-        // if defined scale it, otherwise remain undefined
-        value: v.value && v.value * (s.props.scale ?? 1),
-      }));
+      const transform = s.props.transform ?? (d => d);
+      const scale = s.props.scale ?? 1;
+      const scaledAndTransformedValues = transform(result.datapoints).map(
+        v => ({
+          ...v,
+          // if defined scale/transform it, otherwise remain undefined
+          value: v.value && scale * v.value,
+        }),
+      );
 
       formattedData.push({
-        values: scaledValues,
+        values: scaledAndTransformedValues,
         key: s.props.title || s.props.name,
         area: true,
         fillOpacity: 0.1,

pkg/ui/workspaces/db-console/src/views/shared/components/metricQuery/index.tsx

@@ -40,6 +40,7 @@ import { History } from "history";
 import { TimeWindow } from "src/redux/timeScale";
 import { PayloadAction } from "src/interfaces/action";
 import { AxisUnits, TimeScale } from "@cockroachlabs/cluster-ui";
+import { cockroach } from "@cockroachlabs/crdb-protobuf-client-ccl";
 
 /**
  * AxisProps represents the properties of an Axis being specified as part of a
@@ -97,6 +98,12 @@ export interface MetricProps {
   // to convert it to our Duration format which assumes Nanoseconds.
   scale?: number;
 
+  // Transform is a function that can be applied to the datapoints of the metric
+  // and applies BEFORE scaling
+  transform?: (
+    d: cockroach.ts.tspb.TimeSeriesQueryResponse.IResult["datapoints"],
+  ) => cockroach.ts.tspb.TimeSeriesQueryResponse.IResult["datapoints"];
+
   nonNegativeRate?: boolean;
   aggregateMax?: boolean;
   aggregateMin?: boolean;