[MONIT-29250] Include stack traces in run() methods

src/main/java/com/wavefront/sdk/common/clients/WavefrontClient.java

@@ -619,7 +619,7 @@ public void run() {
       this.flush();
     } catch (Throwable ex) {
       logger.log(LogMessageType.FLUSH_ERROR.toString(), Level.WARNING,
-          "Unable to report to Wavefront cluster: " + Throwables.getRootCause(ex));
+          "Unable to report to Wavefront cluster", Throwables.getRootCause(ex));
     }
   }
 

src/main/java/com/wavefront/sdk/common/logging/MessageDedupingLogger.java

@@ -68,4 +68,26 @@ public void log(String messageDedupingKey, Level level, String message) {
       log(new LogRecord(level, message));
     }
   }
+
+  /**
+   * Log a message, de-duplicating with the specified key.
+   *
+   * @param messageDedupingKey  String to dedupe the log by.
+   * @param level               Log level.
+   * @param message             String to write to log.
+   * @param thrown              Throwable associated with log message.
+   */
+  public void log(String messageDedupingKey, Level level, String message, Throwable thrown) {
+    LogRecord logRecord = new LogRecord(level, message);
+    logRecord.setThrown(thrown);
+    try {
+      if (Objects.requireNonNull(rateLimiterCache.get(messageDedupingKey)).tryAcquire()) {
+        log(logRecord);
+      }
+    } catch (ExecutionException e) {
+      // Log the message if we encounter an error fetching the rate limiter
+      log(logRecord);
+    }
+
+  }
 }

src/main/java/com/wavefront/sdk/direct/ingestion/WavefrontDirectIngestionClient.java

@@ -401,7 +401,7 @@ public void run() {
       this.flush();
     } catch (Throwable ex) {
       logger.log(LogMessageType.FLUSH_ERROR.toString(), Level.WARNING,
-          "Unable to report to Wavefront cluster: " + Throwables.getRootCause(ex));
+          "Unable to report to Wavefront cluster", Throwables.getRootCause(ex));
     }
   }
 

src/test/java/com/wavefront/sdk/common/logging/MessageDedupingLoggerTest.java

@@ -31,12 +31,14 @@ public void testLogger() {
     expect(mockLogger.getName()).andReturn("loggerName").anyTimes();
     replay(mockLogger);
     MessageDedupingLogger log = new MessageDedupingLogger(mockLogger, 1000, 0.1);
+
     reset(mockLogger);
     expect(mockLogger.getName()).andReturn("loggerName").anyTimes();
     Capture<LogRecord> logs = Capture.newInstance(CaptureType.ALL);
     mockLogger.log(capture(logs));
     expectLastCall().times(5);
     replay(mockLogger);
+
     log.severe("msg1");
     log.severe("msg1");
     log.warning("msg1");
@@ -52,6 +54,7 @@ public void testLogger() {
     log.log("msg3", Level.FINE, "message 3");
     log.log("msg5", Level.FINE, "message 5");
     log.log("msg5", Level.FINE, "message 5!");
+
     verify(mockLogger);
     assertEquals(5, logs.getValues().size());
     assertEquals("msg1", logs.getValues().get(0).getMessage());
@@ -64,4 +67,38 @@ public void testLogger() {
     assertEquals(Level.SEVERE, logs.getValues().get(3).getLevel());
     assertEquals("message 5", logs.getValues().get(4).getMessage());
   }
+
+  @Test
+  public void testLoggingThrowables() {
+    Logger mockLogger = EasyMock.mock(Logger.class);
+    expect(mockLogger.getName()).andReturn("loggerName").anyTimes();
+    replay(mockLogger);
+    MessageDedupingLogger logger = new MessageDedupingLogger(mockLogger, 1000, 0.1);
+
+    reset(mockLogger);
+    expect(mockLogger.getName()).andReturn("loggerName").anyTimes();
+    Capture<LogRecord> capturedLogs = Capture.newInstance(CaptureType.ALL);
+    mockLogger.log(capture(capturedLogs));
+    expectLastCall().times(2);
+    replay(mockLogger);
+
+    Exception testException = new Exception("testing");
+    logger.log("key1", Level.WARNING, "error", testException);
+    logger.log("key1", Level.SEVERE, "error", testException);
+    logger.log("key2", Level.SEVERE, "error", testException);
+
+    verify(mockLogger);
+    assertEquals(2, capturedLogs.getValues().size());
+
+    LogRecord capLog1 = capturedLogs.getValues().get(0);
+    assertEquals("loggerName", capLog1.getLoggerName());
+    assertEquals("error", capLog1.getMessage());
+    assertEquals(Level.WARNING, capLog1.getLevel());
+
+    LogRecord capLog2 = capturedLogs.getValues().get(1);
+    assertEquals("loggerName", capLog2.getLoggerName());
+    assertEquals("error", capLog2.getMessage());
+    assertEquals(Level.SEVERE, capLog2.getLevel());
+    assertEquals(testException, capLog2.getThrown());
+  }
 }