Add Embedded mode

src/main/java/org/datadog/jmxfetch/App.java

@@ -92,15 +92,16 @@ public class App {
     public App(AppConfig appConfig) {
         this.appConfig = appConfig;
 
-        ExecutorService collectionThreadPool =
-                buildExecutorService(appConfig.getThreadPoolSize(), COLLECTION_POOL_NAME);
-        collectionProcessor =
-                new TaskProcessor(collectionThreadPool, appConfig.getReporter());
-
-        ExecutorService recoveryThreadPool =
-                buildExecutorService(appConfig.getReconnectionThreadPoolSize(), RECOVERY_POOL_NAME);
+        ExecutorService collectionThreadPool = null;
+        ExecutorService recoveryThreadPool = null;
+        if (!appConfig.isEmbedded()) { // Creates executors in standalone mode only
+            collectionThreadPool = buildExecutorService(appConfig.getThreadPoolSize(),
+                    COLLECTION_POOL_NAME);
+            recoveryThreadPool = buildExecutorService(appConfig.getReconnectionThreadPoolSize(),
+                    RECOVERY_POOL_NAME);
+        }
         recoveryProcessor = new TaskProcessor(recoveryThreadPool, appConfig.getReporter());
-
+        collectionProcessor = new TaskProcessor(collectionThreadPool, appConfig.getReporter());
         // setup client
         if (appConfig.remoteEnabled()) {
             client = new HttpClient(appConfig.getIpcHost(), appConfig.getIpcPort(), false);
@@ -1127,9 +1128,9 @@ private <T> void processCollectionStatus(
             String scStatus = Status.STATUS_OK;
             LinkedList<HashMap<String, Object>> metrics;
 
-            Integer numberOfMetrics = new Integer(0);
+            int numberOfMetrics = 0;
 
-            InstanceTask task = tasks.get(i);
+            InstanceTask<T> task = tasks.get(i);
             TaskStatusHandler status = statuses.get(i);
             Instance instance = task.getInstance();
             Reporter reporter = appConfig.getReporter();
@@ -1195,7 +1196,7 @@ private <T> void processCollectionStatus(
                         appConfig,
                         reporter,
                         instance,
-                        numberOfMetrics.intValue(),
+                        numberOfMetrics,
                         instanceMessage,
                         instanceStatus);
                 this.sendServiceCheck(reporter, instance, instanceMessage, scStatus);

src/main/java/org/datadog/jmxfetch/AppConfig.java

@@ -220,6 +220,14 @@ public class AppConfig {
     @Builder.Default
     private boolean daemon = false;
 
+    /**
+     * Boolean setting to determine whether JMXFetch is embedded in a client app, e.g. for the java
+     * tracer. This setting is uncoupled from daemon one, even though very similar. This setting
+     * is used to reduce number of threads used by assuming the JMX connection will be local.
+     */
+    @Builder.Default
+    private boolean embedded = false;
+
     // This is used by things like APM agent to provide configuration from resources
     private List<String> instanceConfigResources;
     // This is used by things like APM agent to provide metric configuration from resources
@@ -389,4 +397,8 @@ public Map<String, String> getGlobalTags() {
     public boolean isDaemon() {
         return daemon;
     }
+
+    public boolean isEmbedded() {
+        return embedded;
+    }
 }

src/main/java/org/datadog/jmxfetch/tasks/TaskProcessor.java

@@ -36,6 +36,10 @@ public void setThreadPoolExecutor(ExecutorService executor) {
      * for work on tasks.
      * */
     public boolean ready() {
+        if (threadPoolExecutor == null) {
+            // assumes we are in embedded mode and tasks will process by the calling thread
+            return true;
+        }
         ThreadPoolExecutor tpe = (ThreadPoolExecutor) threadPoolExecutor;
         return !tpe.isTerminated() && !(tpe.getMaximumPoolSize() == tpe.getActiveCount());
     }
@@ -46,44 +50,81 @@ public boolean ready() {
     public <T> List<TaskStatusHandler> processTasks(
             List<InstanceTask<T>> tasks, int timeout, TimeUnit timeUnit, TaskMethod<T> processor)
             throws Exception {
-
         List<TaskStatusHandler> statuses = new ArrayList<TaskStatusHandler>();
-
         try {
-            List<Callable<T>> callables = new ArrayList<Callable<T>>();
-            for (InstanceTask<T> task : tasks) {
-                callables.add(task);
-            }
-            List<Future<T>> results = threadPoolExecutor.invokeAll(callables, timeout, timeUnit);
-
-            for (int i = 0; i < results.size(); i++) {
-
-                Instance instance = tasks.get(i).getInstance();
-                try {
-                    Future<T> future = results.get(i);
-
-                    statuses.add(processor.invoke(instance, future, reporter));
-
-                } catch (Exception e) {
-                    log.warn(
-                            "There was an error processing concurrent instance: " + instance, e);
-
-                    statuses.add(new TaskStatusHandler(e));
+            if (threadPoolExecutor != null) {
+                List<Callable<T>> callables = new ArrayList<Callable<T>>(tasks);
+                List<Future<T>> results = threadPoolExecutor.invokeAll(callables, timeout,
+                        timeUnit);
+                for (int i = 0; i < results.size(); i++) {
+                    Instance instance = tasks.get(i).getInstance();
+                    try {
+                        Future<T> future = results.get(i);
+                        statuses.add(processor.invoke(instance, future, reporter));
+                    } catch (Exception e) {
+                        log.warn("There was an error processing concurrent instance: "
+                                        + instance, e);
+                        statuses.add(new TaskStatusHandler(e));
+                    }
+                }
+            } else {
+                for (InstanceTask<T> task : tasks) {
+                    T result = task.call();
+                    statuses.add(processor.invoke(task.getInstance(), new SimpleFuture<T>(result),
+                            reporter));
                 }
             }
+
         } catch (Exception e) {
             // Should we do anything else here?
             log.warn("JMXFetch internal TaskProcessor error invoking concurrent tasks: ", e);
             throw e;
         }
-
         return statuses;
     }
 
     /**
      * Stops the excutor service.
      * */
     public void stop() {
-        threadPoolExecutor.shutdownNow();
+        if (threadPoolExecutor != null) {
+            threadPoolExecutor.shutdownNow();
+        }
+    }
+
+    /**
+     * used to wrap the result in embedded mode when not using executors.
+     */
+    private static class SimpleFuture<T> implements Future<T> {
+        private final T result;
+
+        public SimpleFuture(T result) {
+            this.result = result;
+        }
+
+        @Override
+        public boolean cancel(boolean mayInterruptIfRunning) {
+            return false;
+        }
+
+        @Override
+        public boolean isCancelled() {
+            return false;
+        }
+
+        @Override
+        public boolean isDone() {
+            return true;
+        }
+
+        @Override
+        public T get() {
+            return result;
+        }
+
+        @Override
+        public T get(long timeout, TimeUnit unit) {
+            return result;
+        }
     }
 }

src/test/java/org/datadog/jmxfetch/tasks/TestTaskProcessor.java

@@ -100,17 +100,7 @@ public TaskStatusHandler invoke(Instance instance, Future<Boolean> future, Repor
                         return TestTaskProcessor.processTestResults(instance, future, reporter);
                     };
                 });
-
-        // this should all be green
-        for (int i=0 ; i<statuses.size(); i++) {
-
-            TaskStatusHandler status = statuses.get(i);
-
-            status.raiseForStatus();
-
-            // It should be true - both instances ready to collect
-            assertTrue((Boolean)status.getData());
-        }
+        assertAllStatusGreen(statuses);
     }
 
     /**
@@ -155,4 +145,32 @@ public TaskStatusHandler invoke(Instance instance, Future<Boolean> future, Repor
             }
         }
     }
+
+    @Test
+    public void embeddedTaskProcessor() throws Throwable {
+        // null executor means embedded mode
+        TaskProcessor testProcessor = new TaskProcessor(null, null);
+        List<InstanceTask<Boolean>> instanceTestTasks = new ArrayList<InstanceTask<Boolean>>();
+        for (Instance instance: instances) {
+            instanceTestTasks.add(new TestSimpleTask(instance));
+        }
+        List<TaskStatusHandler> statuses = testProcessor.processTasks(
+                instanceTestTasks, 0, TimeUnit.SECONDS,
+                new TaskMethod<Boolean>() {
+                    @Override
+                    public TaskStatusHandler invoke(Instance instance, Future<Boolean> future, Reporter reporter) {
+                        return TestTaskProcessor.processTestResults(instance, future, reporter);
+                    };
+                });
+        assertTrue(statuses.size() > 0);
+        assertAllStatusGreen(statuses);
+
+    }
+
+    private void assertAllStatusGreen(List<TaskStatusHandler> statuses) throws Throwable {
+        for (TaskStatusHandler status : statuses) {
+            status.raiseForStatus();
+            assertTrue((Boolean)status.getData());
+        }
+    }
 }