Add MosApiMetrics exporter (#2931)

* Add MosApiMetrics exporter with status code mapping

Introduces the metrics exporter for the MoSAPI system.

- Implements `MosApiMetrics` to export TLD and service states to Cloud Monitoring.
- Maps ICANN status codes to numeric gauges: 1 (UP), 0 (DOWN), and 2 (DISABLED/INCONCLUSIVE).
- Sets `MAX_TIMESERIES_PER_REQUEST` to 195 to respect Cloud Monitoring API limits

* Automate metric descriptor creation on startup in Cloud Monitoring

* Refactor MoSAPI metrics for resilience and standards

* Refactor and nits

- Kept projectName as part constant instead of inside method signature
- Added Summary logs for metrics execution
- Metric Executor defaults to Single Threaded

* junit test refactoring

* Fix Metric kind to GAUGE for all metrics

* Refactor MosApiMetrics to remove async ExecutorService

* Add LockHandler for Metric Descriptor creation

* Update LockHandler lease time to one hour and refactoring

core/src/main/java/google/registry/config/RegistryConfig.java

@@ -1463,9 +1463,9 @@ public final class RegistryConfig {
     }
 
     @Provides
-    @Config("mosapiTldThreadCnt")
+    @Config("mosapiTldThreadCount")
     public static int provideMosapiTldThreads(RegistryConfigSettings config) {
-      return config.mosapi.tldThreadCnt;
+      return config.mosapi.tldThreadCount;
     }
 
     private static String formatComments(String text) {

core/src/main/java/google/registry/config/RegistryConfigSettings.java

@@ -272,6 +272,6 @@ public class RegistryConfigSettings {
     public String entityType;
     public List<String> tlds;
     public List<String> services;
-    public int tldThreadCnt;
+    public int tldThreadCount;
   }
 }

core/src/main/java/google/registry/config/files/default-config.yaml

@@ -645,5 +645,5 @@ mosapi:
   # Provides a fixed thread pool for parallel TLD processing.
   # @see <a href="https://www.icann.org/mosapi-specification.pdf">
   # ICANN MoSAPI Specification, Section 12.3</a>
-  tldThreadCnt: 4
+  tldThreadCount: 4
 

core/src/main/java/google/registry/mosapi/MosApiMetrics.java

@@ -14,21 +14,346 @@
 
 package google.registry.mosapi;
 
+import static com.google.common.collect.ImmutableList.toImmutableList;
+
+import com.google.api.client.googleapis.json.GoogleJsonResponseException;
+import com.google.api.services.monitoring.v3.Monitoring;
+import com.google.api.services.monitoring.v3.model.CreateTimeSeriesRequest;
+import com.google.api.services.monitoring.v3.model.LabelDescriptor;
+import com.google.api.services.monitoring.v3.model.Metric;
+import com.google.api.services.monitoring.v3.model.MetricDescriptor;
+import com.google.api.services.monitoring.v3.model.MonitoredResource;
+import com.google.api.services.monitoring.v3.model.Point;
+import com.google.api.services.monitoring.v3.model.TimeInterval;
+import com.google.api.services.monitoring.v3.model.TimeSeries;
+import com.google.api.services.monitoring.v3.model.TypedValue;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Ascii;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Iterators;
 import com.google.common.flogger.FluentLogger;
+import google.registry.config.RegistryConfig.Config;
+import google.registry.mosapi.MosApiModels.ServiceStatus;
 import google.registry.mosapi.MosApiModels.TldServiceState;
+import google.registry.request.lock.LockHandler;
+import google.registry.util.Clock;
 import jakarta.inject.Inject;
+import java.io.IOException;
+import java.time.Instant;
+import java.util.Iterator;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Stream;
+import org.joda.time.Duration;
 
 /** Metrics Exporter for MoSAPI. */
 public class MosApiMetrics {
 
   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
 
-  @Inject
-  public MosApiMetrics() {}
+  // Google Cloud Monitoring Limit: Max 200 TimeSeries per request
+  private static final int MAX_TIMESERIES_PER_REQUEST = 195;
 
-  public void recordStates(List<TldServiceState> states) {
-    // b/467541269: Logic to push status to Cloud Monitoring goes here
-    logger.atInfo().log("MoSAPI record metrics logic will be implemented from here");
+  private static final int METRICS_ALREADY_EXIST = 409;
+
+  // Magic String Constants
+  private static final String METRIC_DOMAIN = "custom.googleapis.com/mosapi/";
+  private static final String PROJECT_RESOURCE_PREFIX = "projects/";
+  private static final String RESOURCE_TYPE_GLOBAL = "global";
+  private static final String LABEL_PROJECT_ID = "project_id";
+  private static final String LABEL_TLD = "tld";
+  private static final String LABEL_SERVICE_TYPE = "service_type";
+
+  // Lock Constants
+  private static final String LOCK_NAME = "MosApiMetricCreation";
+  private static final Duration LOCK_LEASE_TIME = Duration.standardHours(1);
+  // Metric Names
+  private static final String METRIC_TLD_STATUS = "tld_status";
+  private static final String METRIC_SERVICE_STATUS = "service_status";
+  private static final String METRIC_EMERGENCY_USAGE = "emergency_usage";
+  private static final String GAUGE_METRIC_KIND = "GAUGE";
+
+  // Metric Display Names & Descriptions
+  private static final String DISPLAY_NAME_TLD_STATUS =
+      "Health of TLDs. 1 = UP, 0 = DOWN,  2= DISABLED/NOT_MONITORED";
+  private static final String DESC_TLD_STATUS = "Overall Health of TLDs reported from ICANN";
+
+  private static final String DISPLAY_NAME_SERVICE_STATUS =
+      "Health of Services. 1 = UP, 0 = DOWN,  2= DISABLED/NOT_MONITORED";
+  private static final String DESC_SERVICE_STATUS =
+      "Overall Health of Services reported from ICANN";
+
+  private static final String DISPLAY_NAME_EMERGENCY_USAGE =
+      "Percentage of Emergency Threshold Consumed";
+  private static final String DESC_EMERGENCY_USAGE =
+      "Downtime threshold that if reached by any of the monitored Services may cause the TLDs"
+          + " Services emergency transition to an interim Registry Operator";
+
+  // MoSAPI Status Constants
+  private static final String STATUS_UP_INCONCLUSIVE = "UP-INCONCLUSIVE";
+  private static final String STATUS_DOWN = "DOWN";
+  private static final String STATUS_DISABLED = "DISABLED";
+
+  private final Monitoring monitoringClient;
+  private final String projectId;
+  private final String projectName;
+  private final Clock clock;
+  private final MonitoredResource monitoredResource;
+  private final LockHandler lockHandler;
+  // Flag to ensure we only create descriptors once, lazily
+  @VisibleForTesting static final AtomicBoolean isDescriptorInitialized = new AtomicBoolean(false);
+
+  @Inject
+  public MosApiMetrics(
+      Monitoring monitoringClient,
+      @Config("projectId") String projectId,
+      Clock clock,
+      LockHandler lockHandler) {
+    this.monitoringClient = monitoringClient;
+    this.projectId = projectId;
+    this.clock = clock;
+    this.projectName = PROJECT_RESOURCE_PREFIX + projectId;
+    this.lockHandler = lockHandler;
+    this.monitoredResource =
+        new MonitoredResource()
+            .setType(RESOURCE_TYPE_GLOBAL)
+            .setLabels(ImmutableMap.of(LABEL_PROJECT_ID, projectId));
+  }
+
+  /** Accepts a list of states and processes them in a single async batch task. */
+  public void recordStates(ImmutableList<TldServiceState> states) {
+    // If this is the first time we are recording, ensure descriptors exist.
+    ensureMetricDescriptorsWithLock();
+    pushBatchMetrics(states);
+  }
+
+  /**
+   * Attempts to create metric descriptors using a distributed lock.
+   *
+   * <p>If the lock is acquired, this instance creates the descriptors and marks itself initialized.
+   * If the lock is busy, it implies another instance is handling it, so we skip and proceed.
+   */
+  private void ensureMetricDescriptorsWithLock() {
+    lockHandler.executeWithLocks(
+        () -> {
+          if (!isDescriptorInitialized.get()) {
+            createCustomMetricDescriptors();
+            isDescriptorInitialized.set(true);
+          }
+          return null;
+        },
+        null,
+        LOCK_LEASE_TIME,
+        LOCK_NAME);
+  }
+
+  // Defines the custom metrics in Cloud Monitoring
+  private void createCustomMetricDescriptors() {
+    // 1. TLD Status Descriptor
+    createMetricDescriptor(
+        METRIC_TLD_STATUS,
+        DISPLAY_NAME_TLD_STATUS,
+        DESC_TLD_STATUS,
+        "INT64",
+        ImmutableList.of(LABEL_TLD));
+
+    // 2. Service Status Descriptor
+    createMetricDescriptor(
+        METRIC_SERVICE_STATUS,
+        DISPLAY_NAME_SERVICE_STATUS,
+        DESC_SERVICE_STATUS,
+        "INT64",
+        ImmutableList.of(LABEL_TLD, LABEL_SERVICE_TYPE));
+
+    // 3. Emergency Usage Descriptor
+    createMetricDescriptor(
+        METRIC_EMERGENCY_USAGE,
+        DISPLAY_NAME_EMERGENCY_USAGE,
+        DESC_EMERGENCY_USAGE,
+        "DOUBLE",
+        ImmutableList.of(LABEL_TLD, LABEL_SERVICE_TYPE));
+
+    logger.atInfo().log("Metric descriptors ensured for project %s", projectId);
+  }
+
+  private void createMetricDescriptor(
+      String metricTypeSuffix,
+      String displayName,
+      String description,
+      String valueType,
+      ImmutableList<String> labelKeys) {
+
+    ImmutableList<LabelDescriptor> labelDescriptors =
+        labelKeys.stream()
+            .map(
+                key ->
+                    new LabelDescriptor()
+                        .setKey(key)
+                        .setValueType("STRING")
+                        .setDescription(
+                            key.equals(LABEL_TLD)
+                                ? "The TLD being monitored"
+                                : "The type of service"))
+            .collect(toImmutableList());
+
+    MetricDescriptor descriptor =
+        new MetricDescriptor()
+            .setType(METRIC_DOMAIN + metricTypeSuffix)
+            .setMetricKind(GAUGE_METRIC_KIND)
+            .setValueType(valueType)
+            .setDisplayName(displayName)
+            .setDescription(description)
+            .setLabels(labelDescriptors);
+    try {
+      monitoringClient
+          .projects()
+          .metricDescriptors()
+          .create(this.projectName, descriptor)
+          .execute();
+    } catch (GoogleJsonResponseException e) {
+      if (e.getStatusCode() == METRICS_ALREADY_EXIST) {
+        // the metric already exists. This is expected.
+        logger.atFine().log("Metric descriptor %s already exists.", metricTypeSuffix);
+      } else {
+        logger.atWarning().withCause(e).log(
+            "Failed to create metric descriptor %s. Status: %d",
+            metricTypeSuffix, e.getStatusCode());
+      }
+    } catch (Exception e) {
+      logger.atWarning().withCause(e).log(
+          "Unexpected error creating metric descriptor %s.", metricTypeSuffix);
+    }
+  }
+
+  private void pushBatchMetrics(ImmutableList<TldServiceState> states) {
+    Instant now = Instant.ofEpochMilli(clock.nowUtc().getMillis());
+    TimeInterval interval = new TimeInterval().setEndTime(now.toString());
+    Stream<TimeSeries> allTimeSeriesStream =
+        states.stream().flatMap(state -> createMetricsForState(state, interval));
+
+    Iterator<List<TimeSeries>> batchIterator =
+        Iterators.partition(allTimeSeriesStream.iterator(), MAX_TIMESERIES_PER_REQUEST);
+
+    int successCount = 0;
+    int failureCount = 0;
+
+    // Iterate and count
+    while (batchIterator.hasNext()) {
+      List<TimeSeries> batch = batchIterator.next();
+      try {
+        CreateTimeSeriesRequest request = new CreateTimeSeriesRequest().setTimeSeries(batch);
+        monitoringClient.projects().timeSeries().create(this.projectName, request).execute();
+
+        successCount++;
+
+      } catch (IOException e) {
+        failureCount++;
+        // Log individual batch failures, so we have the stack trace for debugging
+        logger.atWarning().withCause(e).log(
+            "Failed to push batch of %d time series.", batch.size());
+      }
+    }
+
+    // 4. Log the final summary
+    if (failureCount > 0) {
+      logger.atWarning().log(
+          "Metric push finished with errors. Batches Succeeded: %d, Failed: %d",
+          successCount, failureCount);
+    } else {
+      logger.atInfo().log("Metric push finished successfully. Batches Succeeded: %d", successCount);
+    }
+  }
+
+  /** Generates all TimeSeries (TLD + Services) for a single state object. */
+  private Stream<TimeSeries> createMetricsForState(TldServiceState state, TimeInterval interval) {
+    // 1. TLD Status
+    Stream<TimeSeries> tldStream = Stream.of(createTldStatusTimeSeries(state, interval));
+
+    // 2. Service Metrics (if any)
+    Stream<TimeSeries> serviceStream =
+        state.serviceStatuses().entrySet().stream()
+            .flatMap(
+                entry ->
+                    createServiceMetricsStream(
+                        state.tld(), entry.getKey(), entry.getValue(), interval));
+
+    return Stream.concat(tldStream, serviceStream);
+  }
+
+  private Stream<TimeSeries> createServiceMetricsStream(
+      String tld, String serviceType, ServiceStatus statusObj, TimeInterval interval) {
+    ImmutableMap<String, String> labels =
+        ImmutableMap.of(LABEL_TLD, tld, LABEL_SERVICE_TYPE, serviceType);
+
+    return Stream.of(
+        createTimeSeries(
+            METRIC_SERVICE_STATUS, labels, parseServiceStatus(statusObj.status()), interval),
+        createTimeSeries(METRIC_EMERGENCY_USAGE, labels, statusObj.emergencyThreshold(), interval));
+  }
+
+  private TimeSeries createTldStatusTimeSeries(TldServiceState state, TimeInterval interval) {
+    return createTimeSeries(
+        METRIC_TLD_STATUS,
+        ImmutableMap.of(LABEL_TLD, state.tld()),
+        parseTldStatus(state.status()),
+        interval);
+  }
+
+  private TimeSeries createTimeSeries(
+      String suffix, ImmutableMap<String, String> labels, Number val, TimeInterval interval) {
+    Metric metric = new Metric().setType(METRIC_DOMAIN + suffix).setLabels(labels);
+
+    TypedValue tv = new TypedValue();
+    if (val instanceof Double) {
+      tv.setDoubleValue((Double) val);
+    } else {
+      tv.setInt64Value(val.longValue());
+    }
+
+    return new TimeSeries()
+        .setMetric(metric)
+        .setResource(this.monitoredResource)
+        .setPoints(ImmutableList.of(new Point().setInterval(interval).setValue(tv)));
+  }
+
+  /**
+   * Translates MoSAPI status to a numeric metric.
+   *
+   * <p>Mappings: 1 (UP) = Healthy; 0 (DOWN) = Critical failure; 2 (UP-INCONCLUSIVE) = Disabled/Not
+   * Monitored/In Maintenance.
+   *
+   * <p>A status of 2 indicates the SLA monitoring system is under maintenance. The TLD is
+   * considered "UP" by default, but individual service checks are disabled. This distinguishes
+   * maintenance windows from actual availability or outages.
+   *
+   * @see <a href="https://www.icann.org/mosapi-specification.pdf">ICANN MoSAPI Spec Sec 5.1</a>
+   */
+  private long parseTldStatus(String status) {
+    return switch (Ascii.toUpperCase(status)) {
+      case STATUS_DOWN -> 0;
+      case STATUS_UP_INCONCLUSIVE -> 2;
+      default -> 1; // status is up
+    };
+  }
+
+  /**
+   * Translates MoSAPI service status to a numeric metric.
+   *
+   * <p>Mappings: 1 (UP) = Healthy; 0 (DOWN) = Critical failure; 2 (DISABLED/UP-INCONCLUSIVE*) =
+   * Disabled/Not Monitored/In Maintenance.
+   *
+   * @see <a href="https://www.icann.org/mosapi-specification.pdf">ICANN MoSAPI Spec Sec 5.1</a>
+   */
+  private long parseServiceStatus(String status) {
+    String serviceStatus = Ascii.toUpperCase(status);
+    if (serviceStatus.startsWith(STATUS_UP_INCONCLUSIVE)) {
+      return 2;
+    }
+    return switch (serviceStatus) {
+      case STATUS_DOWN -> 0;
+      case STATUS_DISABLED -> 2;
+      default -> 1; // status is Up
+    };
   }
 }

core/src/main/java/google/registry/mosapi/MosApiStateService.java

@@ -26,11 +26,9 @@ import google.registry.mosapi.MosApiModels.ServiceStatus;
 import google.registry.mosapi.MosApiModels.TldServiceState;
 import jakarta.inject.Inject;
 import jakarta.inject.Named;
-import java.util.List;
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
-import java.util.stream.Collectors;
 
 /** A service that provides business logic for interacting with MoSAPI Service State. */
 public class MosApiStateService {
@@ -135,11 +133,12 @@ public class MosApiStateService {
                         tldExecutor))
             .collect(toImmutableList());
 
-    List<TldServiceState> allStates =
+    ImmutableList<TldServiceState> allStates =
         futures.stream()
             .map(CompletableFuture::join)
             .filter(Objects::nonNull)
-            .collect(Collectors.toList());
+            .filter(this::isValidForMetrics)
+            .collect(toImmutableList());
 
     if (!allStates.isEmpty()) {
       try {
@@ -152,4 +151,14 @@ public class MosApiStateService {
       logger.atWarning().log("No successful TLD states fetched; skipping metrics push.");
     }
   }
+
+  private boolean isValidForMetrics(TldServiceState state) {
+    if (state.tld() == null || state.status() == null) {
+      logger.atSevere().log(
+          "Contract Violation: Received invalid state (TLD=%s, Status=%s). Skipping.",
+          state.tld(), state.status());
+      return false;
+    }
+    return true;
+  }
 }

core/src/main/java/google/registry/mosapi/module/MosApiModule.java

@@ -200,7 +200,7 @@ public final class MosApiModule {
   @Singleton
   @Named("mosapiTldExecutor")
   static ExecutorService provideMosapiTldExecutor(
-      @Config("mosapiTldThreadCnt") int threadPoolSize) {
+      @Config("mosapiTldThreadCount") int threadPoolSize) {
     return Executors.newFixedThreadPool(threadPoolSize);
   }
 }

core/src/test/java/google/registry/mosapi/MosApiMetricsTest.java

@@ -0,0 +1,236 @@
+// Copyright 2026 The Nomulus Authors. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+package google.registry.mosapi;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import com.google.api.services.monitoring.v3.Monitoring;
+import com.google.api.services.monitoring.v3.model.CreateTimeSeriesRequest;
+import com.google.api.services.monitoring.v3.model.MetricDescriptor;
+import com.google.api.services.monitoring.v3.model.TimeSeries;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import google.registry.mosapi.MosApiModels.ServiceStatus;
+import google.registry.mosapi.MosApiModels.TldServiceState;
+import google.registry.request.lock.LockHandler;
+import google.registry.testing.FakeClock;
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.Callable;
+import org.joda.time.DateTime;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+/** Unit tests for {@link MosApiMetrics}. */
+public class MosApiMetricsTest {
+  private static final String PROJECT_ID = "domain-registry-test";
+
+  private final Monitoring monitoringClient = mock(Monitoring.class);
+  private final Monitoring.Projects projects = mock(Monitoring.Projects.class);
+  private final LockHandler lockHandler = mock(LockHandler.class);
+  private final Monitoring.Projects.TimeSeries timeSeriesResource =
+      mock(Monitoring.Projects.TimeSeries.class);
+  private final Monitoring.Projects.TimeSeries.Create createRequest =
+      mock(Monitoring.Projects.TimeSeries.Create.class);
+
+  // Mocks for Metric Descriptors
+  private final Monitoring.Projects.MetricDescriptors metricDescriptorsResource =
+      mock(Monitoring.Projects.MetricDescriptors.class);
+  private final Monitoring.Projects.MetricDescriptors.Create createDescriptorRequest =
+      mock(Monitoring.Projects.MetricDescriptors.Create.class);
+
+  // Fixed Clock for deterministic testing
+  private final FakeClock clock = new FakeClock(DateTime.parse("2026-01-01T12:00:00Z"));
+  private MosApiMetrics mosApiMetrics;
+
+  @BeforeEach
+  void setUp() throws IOException, NoSuchFieldException, IllegalAccessException {
+    MosApiMetrics.isDescriptorInitialized.set(false);
+    when(monitoringClient.projects()).thenReturn(projects);
+    when(projects.timeSeries()).thenReturn(timeSeriesResource);
+    when(timeSeriesResource.create(anyString(), any(CreateTimeSeriesRequest.class)))
+        .thenReturn(createRequest);
+
+    // Setup for Metric Descriptors
+    when(projects.metricDescriptors()).thenReturn(metricDescriptorsResource);
+    when(metricDescriptorsResource.create(anyString(), any(MetricDescriptor.class)))
+        .thenReturn(createDescriptorRequest);
+    when(lockHandler.executeWithLocks(any(Callable.class), any(), any(), any()))
+        .thenAnswer(
+            invocation -> {
+              ((Callable<?>) invocation.getArgument(0)).call();
+              return true;
+            });
+    mosApiMetrics = new MosApiMetrics(monitoringClient, PROJECT_ID, clock, lockHandler);
+  }
+
+  @Test
+  void testRecordStates_lazilyInitializesMetricDescriptors() throws IOException {
+    TldServiceState state = createTldState("test.tld", "UP", "UP");
+
+    mosApiMetrics.recordStates(ImmutableList.of(state));
+
+    ArgumentCaptor<MetricDescriptor> captor = ArgumentCaptor.forClass(MetricDescriptor.class);
+
+    verify(metricDescriptorsResource, times(3))
+        .create(eq("projects/" + PROJECT_ID), captor.capture());
+
+    List<MetricDescriptor> descriptors = captor.getAllValues();
+
+    // Verify TLD Status Descriptor
+    MetricDescriptor tldStatus =
+        descriptors.stream()
+            .filter(d -> d.getType().endsWith("tld_status"))
+            .findFirst()
+            .orElseThrow();
+    assertThat(tldStatus.getMetricKind()).isEqualTo("GAUGE");
+    assertThat(tldStatus.getValueType()).isEqualTo("INT64");
+
+    // Verify Service Status Descriptor
+    MetricDescriptor serviceStatus =
+        descriptors.stream()
+            .filter(d -> d.getType().endsWith("service_status"))
+            .findFirst()
+            .orElseThrow();
+    assertThat(serviceStatus.getMetricKind()).isEqualTo("GAUGE");
+    assertThat(serviceStatus.getValueType()).isEqualTo("INT64");
+
+    // Verify Emergency Usage Descriptor
+    MetricDescriptor emergencyUsage =
+        descriptors.stream()
+            .filter(d -> d.getType().endsWith("emergency_usage"))
+            .findFirst()
+            .orElseThrow();
+    assertThat(emergencyUsage.getMetricKind()).isEqualTo("GAUGE");
+    assertThat(emergencyUsage.getValueType()).isEqualTo("DOUBLE");
+  }
+
+  @Test
+  void testRecordStates_mapsStatusesToCorrectValues() throws IOException {
+    TldServiceState stateUp = createTldState("tld-up", "UP", "UP");
+    TldServiceState stateDown = createTldState("tld-down", "DOWN", "DOWN");
+    TldServiceState stateMaint = createTldState("tld-maint", "UP-INCONCLUSIVE", "DISABLED");
+
+    mosApiMetrics.recordStates(ImmutableList.of(stateUp, stateDown, stateMaint));
+
+    ArgumentCaptor<CreateTimeSeriesRequest> captor =
+        ArgumentCaptor.forClass(CreateTimeSeriesRequest.class);
+    verify(timeSeriesResource).create(eq("projects/" + PROJECT_ID), captor.capture());
+
+    List<TimeSeries> pushedSeries = captor.getValue().getTimeSeries();
+
+    // Verify TLD Status Mappings: 1 (UP), 0 (DOWN), 2 (UP-INCONCLUSIVE)
+    assertThat(getValueFor(pushedSeries, "tld-up", "tld_status")).isEqualTo(1);
+    assertThat(getValueFor(pushedSeries, "tld-down", "tld_status")).isEqualTo(0);
+    assertThat(getValueFor(pushedSeries, "tld-maint", "tld_status")).isEqualTo(2);
+
+    // Verify Service Status Mappings: UP -> 1, DOWN -> 0, DISABLED -> 2
+    assertThat(getValueFor(pushedSeries, "tld-up", "service_status")).isEqualTo(1);
+    assertThat(getValueFor(pushedSeries, "tld-down", "service_status")).isEqualTo(0);
+    assertThat(getValueFor(pushedSeries, "tld-maint", "service_status")).isEqualTo(2);
+
+    // 3. Verify Emergency Usage (DOUBLE)
+    assertThat(getValueFor(pushedSeries, "tld-up", "emergency_usage").doubleValue())
+        .isEqualTo(50.0);
+    assertThat(getValueFor(pushedSeries, "tld-down", "emergency_usage").doubleValue())
+        .isEqualTo(50.0);
+    assertThat(getValueFor(pushedSeries, "tld-maint", "emergency_usage").doubleValue())
+        .isEqualTo(50.0);
+  }
+
+  @Test
+  void testRecordStates_partitionsTimeSeries_atLimit() throws IOException {
+    ImmutableList<TldServiceState> largeBatch =
+        java.util.stream.IntStream.range(0, 70)
+            .mapToObj(i -> createTldState("tld-" + i, "UP", "UP"))
+            .collect(ImmutableList.toImmutableList());
+    mosApiMetrics.recordStates(largeBatch);
+
+    verify(timeSeriesResource, times(2))
+        .create(eq("projects/" + PROJECT_ID), any(CreateTimeSeriesRequest.class));
+  }
+
+  @Test
+  void testMetricStructure_containsExpectedLabelsAndResource() throws IOException {
+    TldServiceState state = createTldState("example.tld", "UP", "UP");
+    mosApiMetrics.recordStates(ImmutableList.of(state));
+
+    ArgumentCaptor<CreateTimeSeriesRequest> captor =
+        ArgumentCaptor.forClass(CreateTimeSeriesRequest.class);
+    verify(timeSeriesResource).create(anyString(), captor.capture());
+
+    TimeSeries ts = captor.getValue().getTimeSeries().get(0);
+
+    assertThat(ts.getMetric().getType()).startsWith("custom.googleapis.com/mosapi/");
+    assertThat(ts.getMetric().getLabels()).containsEntry("tld", "example.tld");
+
+    assertThat(ts.getResource().getType()).isEqualTo("global");
+    assertThat(ts.getResource().getLabels()).containsEntry("project_id", PROJECT_ID);
+
+    // Verify that the interval matches our fixed clock
+    assertThat(ts.getPoints().get(0).getInterval().getEndTime()).isEqualTo("2026-01-01T12:00:00Z");
+  }
+
+  /** Extracts the numeric value for a specific TLD and metric type from a list of TimeSeries. */
+  private Number getValueFor(List<TimeSeries> seriesList, String tld, String metricSuffix) {
+    String fullMetric = "custom.googleapis.com/mosapi/" + metricSuffix;
+    return seriesList.stream()
+        .filter(ts -> tld.equals(ts.getMetric().getLabels().get("tld")))
+        .filter(ts -> ts.getMetric().getType().equals(fullMetric))
+        .findFirst()
+        .map(
+            ts -> {
+              Double dVal = ts.getPoints().get(0).getValue().getDoubleValue();
+              if (dVal != null) {
+                return (Number) dVal;
+              }
+              // Fallback to Int64.
+              return (Number) ts.getPoints().get(0).getValue().getInt64Value();
+            })
+        .get();
+  }
+
+  @Test
+  void testRecordStates_skipsInitialization_ifLockNotAcquired() throws IOException {
+    when(lockHandler.executeWithLocks(any(Callable.class), any(), any(), any())).thenReturn(false);
+
+    TldServiceState state = createTldState("test.tld", "UP", "UP");
+    mosApiMetrics.recordStates(ImmutableList.of(state));
+
+    verify(metricDescriptorsResource, never()).create(anyString(), any());
+  }
+
+  /** Mocks a TldServiceState with a single service status. */
+  private TldServiceState createTldState(String tld, String tldStatus, String serviceStatus) {
+    ServiceStatus sStatus = mock(ServiceStatus.class);
+    when(sStatus.status()).thenReturn(serviceStatus);
+    when(sStatus.emergencyThreshold()).thenReturn(50.0);
+
+    TldServiceState state = mock(TldServiceState.class);
+    when(state.tld()).thenReturn(tld);
+    when(state.status()).thenReturn(tldStatus);
+    when(state.serviceStatuses()).thenReturn(ImmutableMap.of("dns", sStatus));
+
+    return state;
+  }
+}

core/src/test/java/google/registry/mosapi/MosApiStateServiceTest.java

@@ -169,4 +169,23 @@ class MosApiStateServiceTest {
                         && states.stream()
                             .anyMatch(s -> s.tld().equals("tld1") && s.status().equals("Up"))));
   }
+
+  @Test
+  void testTriggerMetrics_filtersOutInvalidContractStates() throws Exception {
+    // 1. Valid State
+    TldServiceState validState = new TldServiceState("tld1", 1L, "Up", ImmutableMap.of());
+
+    // 2. Invalid State (Status is NULL)
+    // We instantiate it directly to simulate a bad response object.
+    TldServiceState invalidState = new TldServiceState("tld2", 2L, null, ImmutableMap.of());
+
+    when(client.getTldServiceState("tld1")).thenReturn(validState);
+    when(client.getTldServiceState("tld2")).thenReturn(invalidState);
+
+    service.triggerMetricsForAllServiceStateSummaries();
+
+    // Verify: Only the valid state (tld1) is passed to recordStates
+    verify(metrics)
+        .recordStates(argThat(states -> states.size() == 1 && states.get(0).tld().equals("tld1")));
+  }
 }