[WIP] Populate each service's data separately

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.airbnb.billow</groupId>
     <artifactId>billow</artifactId>
-    <version>2.21</version>
+    <version>2.24</version>
 
     <licenses>
         <license>
@@ -207,8 +207,8 @@
                 <artifactId>maven-compiler-plugin</artifactId>
                 <version>${maven-compiler-plugin.version}</version>
                 <configuration>
-                    <source>1.7</source>
-                    <target>1.7</target>
+                    <source>8</source>
+                    <target>8</target>
                 </configuration>
             </plugin>
         </plugins>

src/main/java/com/airbnb/billow/AWSDatabaseHolder.java

@@ -1,5 +1,6 @@
 package com.airbnb.billow;
 
+import com.google.common.collect.ImmutableList;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 
@@ -124,17 +125,77 @@ public AWSDatabaseHolder(Config config) {
         rebuild();
     }
 
+    /**
+     * Build a fresh version of the DB completely from scratch
+     */
     public void rebuild() {
+        log.info(String.format("Using AWS Account number: '%s'", awsAccountNumber));
         current = new AWSDatabase(
-            ec2Clients,
-            rdsClients,
-            dynamoDBClients,
-            sqsClients,
-            elasticacheClients,
-            elasticsearchClients,
-            iamClient,
-            awsAccountNumber,
-            awsARNPartition);
+                ec2Clients,
+                rdsClients,
+                dynamoDBClients,
+                sqsClients,
+                elasticacheClients,
+                elasticsearchClients,
+                iamClient,
+                awsAccountNumber,
+                awsARNPartition);
+    }
+
+    /**
+     * Incrementally refresh ec2 instance data
+     */
+    public void refreshEc2Instances() {
+        current.refreshEc2Instances(ec2Clients);
+    }
+
+    /**
+     * Incrementally refresh dyanmo data
+     */
+    public void refreshDynamoTables() {
+        current.refreshDynamoTables(dynamoDBClients);
+    }
+
+    /**
+     * Incrementally refresh rds instance data
+     */
+    public void refreshRdsInstances() {
+        current.refreshRdsInstances(rdsClients);
+    }
+
+    /**
+     * Incrementally refresh EC2 security group data
+     */
+    public void refreshEc2SGs() {
+        current.refreshEc2SGs(ec2Clients);
+    }
+
+    /**
+     * Incrementally refresh sqs data
+     */
+    public void refreshSqsQueues() {
+        current.refreshSqsQueues(sqsClients);
+    }
+
+    /**
+     * Incrementally refresh elasticache data
+     */
+    public void refreshElasticacheClusters() {
+        current.refreshElasticacheClusters(elasticacheClients);
+    }
+
+    /**
+     * Incrementally refresh IAM data
+     */
+    public void refreshIamUsers() {
+        current.refreshIamUsers(iamClient);
+    }
+
+    /**
+     * Incrementally refresh elasticsearch data
+     */
+    public void refreshElasticsearchClusters() {
+        current.refreshElasticsearchClusters(elasticsearchClients);
     }
 
     public HealthCheck.Result healthy() {

src/main/java/com/airbnb/billow/Main.java

@@ -1,5 +1,6 @@
 package com.airbnb.billow;
 
+import com.airbnb.billow.jobs.*;
 import com.codahale.metrics.CachedGauge;
 import com.codahale.metrics.Gauge;
 import com.codahale.metrics.MetricRegistry;
@@ -22,6 +23,7 @@
 import org.eclipse.jetty.servlet.ServletHolder;
 import org.quartz.JobDetail;
 import org.quartz.Scheduler;
+import org.quartz.SchedulerException;
 import org.quartz.SimpleTrigger;
 import org.quartz.impl.StdSchedulerFactory;
 
@@ -71,23 +73,21 @@ protected Long loadValue() {
         metricRegistry.register(databaseAgeMetricName, cacheAgeGauge);
 
         final Scheduler scheduler = StdSchedulerFactory.getDefaultScheduler();
-        scheduler.getContext().put(AWSDatabaseHolderRefreshJob.DB_KEY, dbHolder);
-        scheduler.getContext().put(AWSDatabaseHolderRefreshJob.START_COUNTER_KEY, metricRegistry.counter(jobStartMetricName));
-        scheduler.getContext().put(AWSDatabaseHolderRefreshJob.FAILURE_COUNTER_KEY, metricRegistry.counter(jobFailureMetricName));
-        scheduler.getContext().put(AWSDatabaseHolderRefreshJob.SUCCESS_COUNTER_KEY, metricRegistry.counter(jobSuccessMetricName));
+        scheduler.getContext().put(BaseAWSDatabaseHolderRefreshJob.DB_KEY, dbHolder);
+        scheduler.getContext().put(BaseAWSDatabaseHolderRefreshJob.START_COUNTER_KEY, metricRegistry.counter(jobStartMetricName));
+        scheduler.getContext().put(BaseAWSDatabaseHolderRefreshJob.FAILURE_COUNTER_KEY, metricRegistry.counter(jobFailureMetricName));
+        scheduler.getContext().put(BaseAWSDatabaseHolderRefreshJob.SUCCESS_COUNTER_KEY, metricRegistry.counter(jobSuccessMetricName));
         scheduler.start();
 
-        final SimpleTrigger trigger = newTrigger().
-                withIdentity(AWSDatabaseHolderRefreshJob.NAME).
-                startNow().
-                withSchedule(simpleSchedule().withIntervalInMilliseconds(refreshRate).repeatForever()).
-                build();
+        scheduleJob(scheduler, DynamoRefreshJob.class, refreshRate);
+        scheduleJob(scheduler, Ec2InstanceRefreshJob.class, refreshRate);
+        scheduleJob(scheduler, Ec2SGRefreshJob.class, refreshRate);
+        scheduleJob(scheduler, ElasticacheRefreshJob.class, refreshRate);
+        scheduleJob(scheduler, ElasticsearchRefreshJob.class, refreshRate);
+        scheduleJob(scheduler, IamRefreshJob.class, refreshRate);
+        scheduleJob(scheduler, RdsRefreshJob.class, refreshRate);
+        scheduleJob(scheduler, SqsRefreshJob.class, refreshRate);
 
-        final JobDetail jobDetail = newJob(AWSDatabaseHolderRefreshJob.class).
-                withIdentity(AWSDatabaseHolderRefreshJob.NAME).
-                build();
-
-        scheduler.scheduleJob(jobDetail, trigger);
 
         log.info("Creating age health check");
         healthCheckRegistry.register("DB", new HealthCheck() {
@@ -175,4 +175,18 @@ private static void configureConnectors(Server server) {
                 }
         }
     }
+
+    private static void scheduleJob(Scheduler scheduler, Class jobClass, Long refreshRate) throws SchedulerException {
+        final SimpleTrigger trigger = newTrigger().
+                withIdentity(jobClass.toString()).
+                startNow().
+                withSchedule(simpleSchedule().withIntervalInMilliseconds(refreshRate).repeatForever()).
+                build();
+
+        final JobDetail jobDetail = newJob(jobClass).
+                withIdentity(jobClass.toString()).
+                build();
+
+        scheduler.scheduleJob(jobDetail, trigger);
+    }
 }

src/main/java/com/airbnb/billow/TimestampedData.java

@@ -0,0 +1,18 @@
+package com.airbnb.billow;
+
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.function.Supplier;
+
+@Data
+@Slf4j
+public class TimestampedData<T> {
+    private final T data;
+    private final long timestamp;
+
+    public static <T> TimestampedData<T> withTimestamp(Supplier<T> supplier) {
+        long timestamp = System.currentTimeMillis();
+        return new TimestampedData<T>(supplier.get(), timestamp);
+    }
+}

src/main/java/com/airbnb/billow/AWSDatabaseHolderRefreshJob.java → src/main/java/com/airbnb/billow/jobs/BaseAWSDatabaseHolderRefreshJob.java

@@ -1,30 +1,36 @@
-package com.airbnb.billow;
+package com.airbnb.billow.jobs;
 
+import com.airbnb.billow.AWSDatabaseHolder;
 import com.codahale.metrics.Counter;
+import lombok.extern.slf4j.Slf4j;
 import org.quartz.Job;
 import org.quartz.JobExecutionContext;
 import org.quartz.JobExecutionException;
 import org.quartz.SchedulerException;
 
-public class AWSDatabaseHolderRefreshJob implements Job {
+@Slf4j
+public abstract class BaseAWSDatabaseHolderRefreshJob implements Job {
     public static final String DB_KEY              = "db";
     public static final String FAILURE_COUNTER_KEY = "failure_counter";
     public static final String START_COUNTER_KEY   = "start_counter";
     public static final String SUCCESS_COUNTER_KEY = "success_counter";
-    public static final String NAME                = "dbRefresh";
 
     @Override
     public void execute(JobExecutionContext context) throws JobExecutionException {
         try {
             increment(START_COUNTER_KEY, context);
             ((AWSDatabaseHolder) context.getScheduler().getContext().get(DB_KEY)).rebuild();
+            log.info("[job success] {} completed", this.getClass().toString());
             increment(SUCCESS_COUNTER_KEY, context);
         } catch (SchedulerException e) {
+            log.error("[job failure] {} completed", this.getClass().toString());
             increment(FAILURE_COUNTER_KEY, context);
             throw new JobExecutionException(e);
         }
     }
 
+    abstract void refresh(AWSDatabaseHolder dbHolder);
+
     private void increment(String counterName, JobExecutionContext context) {
         try {
             ((Counter) context.getScheduler().getContext().get(counterName)).inc();

src/main/java/com/airbnb/billow/jobs/DynamoRefreshJob.java

@@ -0,0 +1,12 @@
+package com.airbnb.billow.jobs;
+
+import com.airbnb.billow.AWSDatabaseHolder;
+
+public class DynamoRefreshJob extends BaseAWSDatabaseHolderRefreshJob {
+    public static final String NAME = "dynamo_job";
+
+    @Override
+    void refresh(AWSDatabaseHolder dbHolder) {
+        dbHolder.refreshDynamoTables();
+    }
+}

src/main/java/com/airbnb/billow/jobs/Ec2InstanceRefreshJob.java

@@ -0,0 +1,12 @@
+package com.airbnb.billow.jobs;
+
+import com.airbnb.billow.AWSDatabaseHolder;
+
+public class Ec2InstanceRefreshJob extends BaseAWSDatabaseHolderRefreshJob {
+    public static final String NAME = "ec2_instance_job";
+
+    @Override
+    void refresh(AWSDatabaseHolder dbHolder) {
+        dbHolder.refreshEc2Instances();
+    }
+}

src/main/java/com/airbnb/billow/jobs/Ec2SGRefreshJob.java

@@ -0,0 +1,12 @@
+package com.airbnb.billow.jobs;
+
+import com.airbnb.billow.AWSDatabaseHolder;
+
+public class Ec2SGRefreshJob extends BaseAWSDatabaseHolderRefreshJob {
+    public static final String NAME = "ec2_sg_job";
+
+    @Override
+    void refresh(AWSDatabaseHolder dbHolder) {
+        dbHolder.refreshEc2SGs();
+    }
+}

src/main/java/com/airbnb/billow/jobs/ElasticacheRefreshJob.java

@@ -0,0 +1,12 @@
+package com.airbnb.billow.jobs;
+
+import com.airbnb.billow.AWSDatabaseHolder;
+
+public class ElasticacheRefreshJob extends BaseAWSDatabaseHolderRefreshJob {
+    public static final String NAME = "elasticache_job";
+
+    @Override
+    void refresh(AWSDatabaseHolder dbHolder) {
+        dbHolder.refreshElasticacheClusters();
+    }
+}

src/main/java/com/airbnb/billow/jobs/ElasticsearchRefreshJob.java

@@ -0,0 +1,12 @@
+package com.airbnb.billow.jobs;
+
+import com.airbnb.billow.AWSDatabaseHolder;
+
+public class ElasticsearchRefreshJob extends BaseAWSDatabaseHolderRefreshJob {
+    public static final String NAME = "elasticsearch_job";
+
+    @Override
+    void refresh(AWSDatabaseHolder dbHolder) {
+        dbHolder.refreshElasticsearchClusters();
+    }
+}

src/main/java/com/airbnb/billow/jobs/IamRefreshJob.java

@@ -0,0 +1,12 @@
+package com.airbnb.billow.jobs;
+
+import com.airbnb.billow.AWSDatabaseHolder;
+
+public class IamRefreshJob extends BaseAWSDatabaseHolderRefreshJob {
+    public static final String NAME = "iam_job";
+
+    @Override
+    void refresh(AWSDatabaseHolder dbHolder) {
+        dbHolder.refreshIamUsers();
+    }
+}

src/main/java/com/airbnb/billow/jobs/RdsRefreshJob.java

@@ -0,0 +1,12 @@
+package com.airbnb.billow.jobs;
+
+import com.airbnb.billow.AWSDatabaseHolder;
+
+public class RdsRefreshJob extends BaseAWSDatabaseHolderRefreshJob{
+    public static final String NAME = "rds_job";
+
+    @Override
+    void refresh(AWSDatabaseHolder dbHolder) {
+        dbHolder.refreshRdsInstances();
+    }
+}

src/main/java/com/airbnb/billow/jobs/SqsRefreshJob.java

@@ -0,0 +1,12 @@
+package com.airbnb.billow.jobs;
+
+import com.airbnb.billow.AWSDatabaseHolder;
+
+public class SqsRefreshJob extends BaseAWSDatabaseHolderRefreshJob {
+    public static final String NAME = "sqs_job";
+
+    @Override
+    void refresh(AWSDatabaseHolder dbHolder) {
+        dbHolder.refreshSqsQueues();
+    }
+}