Merge pull request #44 from launchdarkly/dr/eventSampling

Add event sampling. Name threads.

Author: drichelson

build.gradle

@@ -6,18 +6,15 @@ apply plugin: 'idea'
 apply plugin: 'com.github.johnrengelman.shadow'
 
 configurations.all {
-    // check for updates every build
+    // check for updates every build for dependencies with: 'changing: true'
     resolutionStrategy.cacheChangingModulesFor 0, 'seconds'
 }
 
 repositories {
-    mavenCentral()
     mavenLocal()
-
     // Before LaunchDarkly release artifacts get synced to Maven Central they are here along with snapshots:
-    maven {
-        url "https://oss.sonatype.org/content/groups/public/"
-    }
+    maven { url "https://oss.sonatype.org/content/groups/public/" }
+    mavenCentral()
 }
 
 allprojects {
@@ -38,7 +35,7 @@ dependencies {
     compile "redis.clients:jedis:2.8.0"
     testCompile "org.easymock:easymock:3.4"
     testCompile 'junit:junit:[4.10,)'
-    testRuntime "org.slf4j:slf4j-simple:1.7.7"
+    testRuntime "ch.qos.logback:logback-classic:1.1.3"
 }
 
 jar {

src/main/java/com/launchdarkly/client/EventProcessor.java

@@ -1,5 +1,6 @@
 package com.launchdarkly.client;
 
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.gson.Gson;
 import org.apache.http.HttpStatus;
 import org.apache.http.client.methods.CloseableHttpResponse;
@@ -14,22 +15,34 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Random;
 import java.util.concurrent.*;
 
 class EventProcessor implements Closeable {
-  private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new DaemonThreadFactory());
+  private final ScheduledExecutorService scheduler;
+  private final Random random = new Random();
   private final BlockingQueue<Event> queue;
   private final String apiKey;
+  private final LDConfig config;
   private final Consumer consumer;
 
   EventProcessor(String apiKey, LDConfig config) {
     this.apiKey = apiKey;
     this.queue = new ArrayBlockingQueue<>(config.capacity);
     this.consumer = new Consumer(config);
+    this.config = config;
+    ThreadFactory threadFactory = new ThreadFactoryBuilder()
+        .setDaemon(true)
+        .setNameFormat("LaunchDarkly-EventProcessor-%d")
+        .build();
+    this.scheduler = Executors.newSingleThreadScheduledExecutor(threadFactory);
     this.scheduler.scheduleAtFixedRate(consumer, 0, config.flushInterval, TimeUnit.SECONDS);
   }
 
   boolean sendEvent(Event e) {
+    if (config.samplingInterval > 0 && random.nextInt(config.samplingInterval) != 0) {
+      return true;
+    }
     return queue.offer(e);
   }
 
@@ -43,18 +56,8 @@ public void flush() {
     this.consumer.flush();
   }
 
-  static class DaemonThreadFactory implements ThreadFactory {
-    public Thread newThread(Runnable r) {
-      Thread thread = new Thread(r);
-      thread.setDaemon(true);
-      return thread;
-    }
-  }
-
   class Consumer implements Runnable {
     private final Logger logger = LoggerFactory.getLogger(Consumer.class);
-
-
     private final CloseableHttpClient client;
     private final LDConfig config;
 
@@ -78,6 +81,7 @@ public void flush() {
     }
 
     private void postEvents(List<Event> events) {
+      logger.debug("Posting " + events.size() + " event(s)..");
       CloseableHttpResponse response = null;
       Gson gson = new Gson();
       String json = gson.toJson(events);
@@ -95,13 +99,11 @@ private void postEvents(List<Event> events) {
         if (status >= 300) {
           if (status == HttpStatus.SC_UNAUTHORIZED) {
             logger.error("Invalid API key");
-          }
-          else {
+          } else {
             logger.error("Unexpected status code: " + status);
           }
-        }
-        else {
-          logger.debug("Successfully processed events");
+        } else {
+          logger.debug("Successfully posted " + events.size() + " event(s).");
         }
       } catch (IOException e) {
         logger.error("Unhandled exception in LaunchDarkly client attempting to connect to URI: " + config.eventsURI, e);
@@ -112,7 +114,6 @@ private void postEvents(List<Event> events) {
           logger.error("Unhandled exception in LaunchDarkly client", e);
         }
       }
-
     }
   }
 }

src/main/java/com/launchdarkly/client/LDConfig.java

@@ -23,6 +23,7 @@ public final class LDConfig {
   private static final int DEFAULT_FLUSH_INTERVAL = 5;
   private static final long DEFAULT_POLLING_INTERVAL_MILLIS = 1000L;
   private static final long DEFAULT_START_WAIT_MILLIS = 0L;
+  private static final int DEFAULT_SAMPLING_INTERVAL = 0;
   private static final Logger logger = LoggerFactory.getLogger(LDConfig.class);
 
   protected static final LDConfig DEFAULT = new Builder().build();
@@ -42,6 +43,7 @@ public final class LDConfig {
   final boolean offline;
   final long pollingIntervalMillis;
   final long startWaitMillis;
+  final int samplingInterval;
 
   protected LDConfig(Builder builder) {
     this.baseURI = builder.baseURI;
@@ -63,6 +65,7 @@ protected LDConfig(Builder builder) {
       this.pollingIntervalMillis = builder.pollingIntervalMillis;
     }
     this.startWaitMillis = builder.startWaitMillis;
+    this.samplingInterval = builder.samplingInterval;
   }
 
   /**
@@ -94,6 +97,7 @@ public static class Builder {
     private long pollingIntervalMillis = DEFAULT_POLLING_INTERVAL_MILLIS;
     private FeatureStore featureStore = new InMemoryFeatureStore();
     public long startWaitMillis = DEFAULT_START_WAIT_MILLIS;
+    public int samplingInterval = DEFAULT_SAMPLING_INTERVAL;
 
     /**
      * Creates a builder with all configuration parameters set to the default
@@ -335,6 +339,21 @@ public Builder startWaitMillis(long startWaitMillis) {
       return this;
     }
 
+   /**
+     * Enable event sampling. When set to the default of zero, sampling is disabled and all events
+     * are sent back to LaunchDarkly. When set to greater than zero, there is a 1 in
+     * <code>samplingInterval</code> chance events will be will be sent.
+     *
+     * <p>Example: if you want 5% sampling rate, set <code>samplingInterval</code> to 20.
+     *
+     * @param samplingInterval the sampling interval.
+     * @return the builder
+     */
+    public Builder samplingInterval(int samplingInterval) {
+      this.samplingInterval = samplingInterval;
+      return this;
+    }
+
     HttpHost proxyHost() {
       if (this.proxyHost == null && this.proxyPort == -1 && this.proxyScheme == null) {
         return null;

src/main/java/com/launchdarkly/client/PollingProcessor.java

@@ -1,13 +1,11 @@
 package com.launchdarkly.client;
 
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 public class PollingProcessor implements UpdateProcessor {
@@ -40,7 +38,10 @@ public Future<Void> start() {
     logger.info("Starting LaunchDarkly polling client with interval: "
         + config.pollingIntervalMillis + " milliseconds");
     final VeryBasicFuture initFuture = new VeryBasicFuture();
-    scheduler = Executors.newScheduledThreadPool(1);
+    ThreadFactory threadFactory = new ThreadFactoryBuilder()
+        .setNameFormat("LaunchDarkly-PollingProcessor-%d")
+        .build();
+    scheduler = Executors.newScheduledThreadPool(1, threadFactory);
 
     scheduler.scheduleAtFixedRate(new Runnable() {
       @Override

src/main/java/com/launchdarkly/client/StreamProcessor.java

@@ -104,11 +104,10 @@ else if (name.equals(INDIRECT_PATCH)) {
 
       @Override
       public void onError(Throwable throwable) {
-        logger.error("Encountered exception in LaunchDarkly client: " + throwable.getMessage());
+        logger.warn("Encountered EventSource error", throwable);
       }
     };
 
-
     es = new EventSource.Builder(handler, URI.create(config.streamURI.toASCIIString() + "/features"))
         .headers(headers)
         .build();

src/test/resources/logback.xml

@@ -0,0 +1,13 @@
+<configuration>
+  <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+    <!-- encoders are assigned the type
+         ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->
+    <encoder>
+      <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36}:%line - %msg%n</pattern>
+    </encoder>
+  </appender>
+
+  <root level="debug">
+    <appender-ref ref="STDOUT" />
+  </root>
+</configuration>