DeploymentMetricsMaintainer extends ControllerMaintainer { @Override protected boolean maintain() { AtomicInteger failures = new AtomicInteger(0); AtomicInteger attempts = new AtomicInteger(0); AtomicReference<Exception> lastException = new AtomicReference<>(null); ForkJoinPool pool = new ForkJoinPool(applicationsToUpdateInParallel); pool.submit(() -> applications.readable().parallelStream().forEach(application -> { for (Instance instance : application.instances().values()) for (Deployment deployment : instance.deployments().values()) { attempts.incrementAndGet(); try { if (deployment.version().getMajor() < 7) continue; var collectedMetrics = controller().metrics().getDeploymentMetrics(instance.id(), deployment.zone()); var now = controller().clock().instant(); applications.lockApplicationIfPresent(application.id(), locked -> { Deployment existingDeployment = locked.get().require(instance.name()).deployments().get(deployment.zone()); if (existingDeployment == null) return; DeploymentMetrics newMetrics = existingDeployment.metrics() .withQueriesPerSecond(collectedMetrics.queriesPerSecond()) .withWritesPerSecond(collectedMetrics.writesPerSecond()) .withDocumentCount(collectedMetrics.documentCount()) .withQueryLatencyMillis(collectedMetrics.queryLatencyMillis()) .withWriteLatencyMillis(collectedMetrics.writeLatencyMillis()) .at(now); applications.store(locked.with(instance.name(), lockedInstance -> lockedInstance.with(existingDeployment.zone(), newMetrics) .recordActivityAt(now, existingDeployment.zone()))); }); } catch (Exception e) { failures.incrementAndGet(); lastException.set(e); } } }) ); pool.shutdown(); try { pool.awaitTermination(30, TimeUnit.MINUTES); if (lastException.get() != null) { log.log(Level.WARNING, String.format("Failed to gather metrics for %d/%d applications. Retrying in %s. Last error: %s", failures.get(), attempts.get(), interval(), Exceptions.toMessageString(lastException.get()))); } } catch (InterruptedException e) { throw new RuntimeException(e); } return lastException.get() == null; } DeploymentMetricsMaintainer(Controller controller, Duration duration);   }
@Test public void updates_metrics() { var application = tester.newDeploymentContext(); application.runJob(JobType.devUsEast1, new ApplicationPackage(new byte[0]), Version.fromString("7.1")); DeploymentMetricsMaintainer maintainer = maintainer(tester.controller()); Supplier<Application> app = application::application; Supplier<Deployment> deployment = () -> application.deployment(ZoneId.from("dev", "us-east-1")); assertEquals(0, app.get().metrics().queryServiceQuality(), 0); assertEquals(0, deployment.get().metrics().documentCount(), 0); assertFalse("No timestamp set", deployment.get().metrics().instant().isPresent()); assertFalse("Never received any queries", deployment.get().activity().lastQueried().isPresent()); assertFalse("Never received any writes", deployment.get().activity().lastWritten().isPresent()); application.runJob(JobType.devUsEast1, new ApplicationPackage(new byte[0]), Version.fromString("6.3.3")); maintainer.maintain(); assertEquals(0, app.get().metrics().queryServiceQuality(), 0); assertEquals(0, deployment.get().metrics().documentCount(), 0); assertFalse("No timestamp set", deployment.get().metrics().instant().isPresent()); assertFalse("Never received any queries", deployment.get().activity().lastQueried().isPresent()); assertFalse("Never received any writes", deployment.get().activity().lastWritten().isPresent()); application.runJob(JobType.devUsEast1, new ApplicationPackage(new byte[0]), Version.fromString("7.5.5")); var metrics0 = Map.of(ClusterMetrics.QUERIES_PER_SECOND, 1D, ClusterMetrics.FEED_PER_SECOND, 2D, ClusterMetrics.DOCUMENT_COUNT, 3D, ClusterMetrics.QUERY_LATENCY, 4D, ClusterMetrics.FEED_LATENCY, 5D); setMetrics(application.application().id().defaultInstance(), metrics0); maintainer.maintain(); Instant t1 = tester.clock().instant().truncatedTo(MILLIS); assertEquals(0.0, app.get().metrics().queryServiceQuality(), Double.MIN_VALUE); assertEquals(0.0, app.get().metrics().writeServiceQuality(), Double.MIN_VALUE); assertEquals(1, deployment.get().metrics().queriesPerSecond(), Double.MIN_VALUE); assertEquals(2, deployment.get().metrics().writesPerSecond(), Double.MIN_VALUE); assertEquals(3, deployment.get().metrics().documentCount(), Double.MIN_VALUE); assertEquals(4, deployment.get().metrics().queryLatencyMillis(), Double.MIN_VALUE); assertEquals(5, deployment.get().metrics().writeLatencyMillis(), Double.MIN_VALUE); assertEquals(t1, deployment.get().metrics().instant().get()); assertEquals(t1, deployment.get().activity().lastQueried().get()); assertEquals(t1, deployment.get().activity().lastWritten().get()); tester.clock().advance(Duration.ofHours(1)); Instant t2 = tester.clock().instant().truncatedTo(MILLIS); maintainer.maintain(); assertEquals(t2, deployment.get().metrics().instant().get()); assertEquals(t2, deployment.get().activity().lastQueried().get()); assertEquals(t2, deployment.get().activity().lastWritten().get()); assertEquals(1, deployment.get().activity().lastQueriesPerSecond().getAsDouble(), Double.MIN_VALUE); assertEquals(2, deployment.get().activity().lastWritesPerSecond().getAsDouble(), Double.MIN_VALUE); tester.clock().advance(Duration.ofHours(1)); Instant t3 = tester.clock().instant().truncatedTo(MILLIS); var metrics1 = new HashMap<>(metrics0); metrics1.put(ClusterMetrics.QUERIES_PER_SECOND, 0D); metrics1.put(ClusterMetrics.FEED_PER_SECOND, 5D); setMetrics(application.application().id().defaultInstance(), metrics1); maintainer.maintain(); assertEquals(t2, deployment.get().activity().lastQueried().get()); assertEquals(t3, deployment.get().activity().lastWritten().get()); assertEquals(1, deployment.get().activity().lastQueriesPerSecond().getAsDouble(), Double.MIN_VALUE); assertEquals(5, deployment.get().activity().lastWritesPerSecond().getAsDouble(), Double.MIN_VALUE); tester.clock().advance(Duration.ofHours(1)); var metrics2 = new HashMap<>(metrics1); metrics2.put(ClusterMetrics.FEED_PER_SECOND, 0D); setMetrics(application.application().id().defaultInstance(), metrics2); maintainer.maintain(); assertEquals(t2, deployment.get().activity().lastQueried().get()); assertEquals(t3, deployment.get().activity().lastWritten().get()); assertEquals(1, deployment.get().activity().lastQueriesPerSecond().getAsDouble(), Double.MIN_VALUE); assertEquals(5, deployment.get().activity().lastWritesPerSecond().getAsDouble(), Double.MIN_VALUE); }