MeteredExecutorServiceWrapper implements ExecutorServiceWrapper { @Override public ScheduledExecutorService wrap( ScheduledExecutorService executor, String name, String... tags) { monitorThreadPoolExecutor(executor, name, tags); return ExecutorServiceMetrics.monitor(meterRegistry, executor, name, Tags.of(tags)); } @Inject  MeteredExecutorServiceWrapper(MeterRegistry meterRegistry); @Override ScheduledExecutorService wrap(
      ScheduledExecutorService executor, String name, String... tags); @Override ExecutorService wrap(ExecutorService executor, String name, String... tags); @Override CronExecutorService wrap(CronExecutorService executor, String name, String... tags);  }
@Test public void shouldRecordExecutorServiceMetrics() throws InterruptedException, TimeoutException, ExecutionException { executor = executorServiceWrapper.wrap( Executors.newSingleThreadExecutor(), MeteredExecutorServiceWrapperTest.class.getName(), "userTagKey", "userTagValue"); CountDownLatch runnableTaskStart = new CountDownLatch(1); Future<?> future = executor.submit(runnableTaskStart::countDown); runnableTaskStart.await(10, TimeUnit.SECONDS); future.get(1, TimeUnit.MINUTES); assertCounter("thread.factory.terminated", 0.0); assertGauge("thread.factory.running", 1.0); assertCounter("thread.factory.created", 1.0); assertCounter("executor.rejected", 0.0); assertGauge("executor.pool.size", 1.0); assertFunctionCounter("executor.completed", 1.0); assertGauge("executor.queued", 0.0); assertTimerCount("executor.idle", 1L); assertGauge("executor.queue.remaining", (double) Integer.MAX_VALUE); assertTimerCount("executor", 1L); assertGauge("executor.active", 0.0); }
@Test public void shouldRecordScheduledExecutorServiceMetrics() throws InterruptedException { executor = executorServiceWrapper.wrap( Executors.newSingleThreadScheduledExecutor(), MeteredExecutorServiceWrapperTest.class.getName(), "userTagKey", "userTagValue"); CountDownLatch runnableTaskStart = new CountDownLatch(1); ((ScheduledExecutorService) executor) .scheduleAtFixedRate(runnableTaskStart::countDown, 0, 100, TimeUnit.SECONDS); runnableTaskStart.await(10, TimeUnit.SECONDS); assertCounter("thread.factory.terminated", 0.0); assertGauge("thread.factory.running", 1.0); assertCounter("thread.factory.created", 1.0); assertCounter("executor.rejected", 0.0); assertGauge("executor.pool.size", 1.0); assertFunctionCounter("executor.completed", 1.0); assertGauge("executor.queued", 1.0); assertTimerCount("executor.idle", 0L); assertGauge("executor.queue.remaining", (double) Integer.MAX_VALUE); assertTimerCount("executor", 1L); assertGauge("executor.active", 0.0); assertCounter("executor.scheduled.once", 0.0); assertCounter("executor.scheduled.repetitively", 1.0); }
@Test public void shouldRecordCronExecutorServiceMetrics() throws InterruptedException, ParseException { CronExecutorService executor = executorServiceWrapper.wrap( new CronThreadPoolExecutor(1), MeteredExecutorServiceWrapperTest.class.getName(), "userTagKey", "userTagValue"); CountDownLatch runnableTaskStart = new CountDownLatch(1); executor.schedule(runnableTaskStart::countDown, new CronExpression(" * * * ? * * *")); runnableTaskStart.await(10, TimeUnit.SECONDS); assertCounter("thread.factory.terminated", 0.0); assertGauge("thread.factory.running", 2.0); assertCounter("thread.factory.created", 2.0); assertCounter("executor.rejected", 0.0); assertGauge("executor.pool.size", 2.0); assertFunctionCounter("executor.completed", 1.0); assertGauge("executor.queued", 1.0); assertTimerCount("executor.idle", 0L); assertTrue( registry .get("executor.queue.remaining") .tag("name", MeteredExecutorServiceWrapperTest.class.getName()) .tags(userTags) .gauge() .value() > 0.0); assertTimerCount("executor", 1L); assertGauge("executor.active", 1.0); assertCounter("executor.scheduled.once", 0.0); assertCounter("executor.scheduled.repetitively", 0.0); assertCounter("executor.scheduled.cron", 1.0); }