Feature/hourly counts #274

maxBruckhaus · 2017-08-10T19:06:37Z

This pull request is for the write path of the traffic management feature. I created a new counter table for hourly bucket counts. There is an accessor for the table which increments the counts. I added code to populate assignment stats into a concurrent hashmap. I have unit tests which ensure that I am incrementing the counts in the hashmap correctly. I have a scheduled executor service which runs every 60 minutes, using the writeCounts method to write the hourly counts to Cassandra.

…counts

…sts, cleanup code

…re/hourly-counts

…service

…re/hourly-counts

maxBruckhaus · 2017-08-10T23:38:08Z

...java/com/intuit/wasabi/assignment/countAggregation/impl/AssignmentsHourlyAggregatorTask.java

+
+    @Inject
+    public AssignmentsHourlyAggregatorTask(AssignmentStats assignmentStats){
+        // TODO: Figure out how to pass the correct AssignmentStats instance from CassandraAssignmentsRepository


I'm in the process of fixing my scheduled executor service

@mooxter Ack on this one

iizrailevsky · 2017-08-11T21:54:13Z

modules/assignment/src/main/java/com/intuit/wasabi/assignment/AssignmentsModule.java

@@ -175,6 +180,30 @@ private void bindMetadataCache(final Properties properties) {
        }
    }

+    private void bindAssignmentsHourlyAggregation() {
+        //This is the assignment aggregation interval.
+        Integer assignmentAggregationIntervalInMinutes = 60;


@mooxter Would it make sense to extract & configure the actual value in a properties (assignment.properties?) file, instead of hardcoding it in code? This is in case you'd need to update it (at run-time?) in the future

iizrailevsky · 2017-08-11T21:56:41Z

modules/assignment/src/main/java/com/intuit/wasabi/assignment/AssignmentsModule.java

+        ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("AssignmentHourlyAggregatorTask-%d").setDaemon(true).build();
+        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1, threadFactory);
+
+        //Bind Scheduled Executor Service


@mooxter Awesome job with inline comments! Very helpful, thanks 👍

iizrailevsky · 2017-08-11T22:00:33Z

...tory-datastax/src/main/java/com/intuit/wasabi/repository/cassandra/impl/AssignmentStats.java

+
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.*;


@mooxter Please don't use .* imports, as this loads all (MANY 😄 ) the classes under java.util. into the JVM. Please import specific classes instead that you're actually using. There's a shortcut to for this in your IDE (IntelliJ?).

@iizrailevsky Ack, I'll remove usage of the .*

iizrailevsky · 2017-08-11T22:02:31Z

...tory-datastax/src/main/java/com/intuit/wasabi/repository/cassandra/impl/AssignmentStats.java

+        }
+    }
+
+    public void incrementCount(Experiment experiment, Assignment assignment) {


@mooxter Please add method JavaDoc to include what the method does, input & output parameters.

@iizrailevsky Ack, will add JavaDoc

iizrailevsky · 2017-08-11T22:05:27Z

...tory-datastax/src/main/java/com/intuit/wasabi/repository/cassandra/impl/AssignmentStats.java

+        Bucket.Label bucketLabel = labelOptional.orElseGet(() -> NULL_LABEL);
+        AtomicInteger oldCount = hourMap.get(ExpBucket.getKey(id, bucketLabel));
+        if (oldCount == null) {
+            synchronized (lock) {


@mooxter Mind adding a comment to describe why you need to synchronize here? What's the specific reason?

@iizrailevsky This is to ensure that multiple threads do not change the oldCount value at the same time, skewing the final value. Without this, oldCount would be initialized to 1 twice instead of properly being set to 1 and then incrementing to 2. I'll add a comment to clarify.

iizrailevsky · 2017-08-11T22:08:02Z

...tory-datastax/src/main/java/com/intuit/wasabi/repository/cassandra/impl/AssignmentStats.java

+        }
+    }
+
+    public int getCount(Experiment experiment, Bucket.Label bucketLabel, int assignmentHour) {


@mooxter Same here - please add method JavaDoc's for all the methods in the class

@iizrailevsky Ack, will add

iizrailevsky · 2017-08-11T22:09:57Z

...tory-datastax/src/main/java/com/intuit/wasabi/repository/cassandra/impl/AssignmentStats.java

+            for (int i = 0; i < count; i++){
+                hourlyBucketCountAccessor.incrementCountBy(experimentID, day, bucketLabel, assignmentHour);
+            }
+            System.out.println(pair.getKey() + " = " + pair.getValue());


@mooxter Please use LOGGER.debug(...) (check other classes), insetad of System.out.println(...)

@iizrailevsky This was mainly for me to see if my code was working properly, I forgot to take this out before committing. I'll use LOGGER.debug(...) in the future though 👍

iizrailevsky · 2017-08-11T22:12:41Z

...tory-datastax/src/main/java/com/intuit/wasabi/repository/cassandra/impl/AssignmentStats.java

+        hourlyCountMap.put(assignmentHour, new ConcurrentHashMap<>());
+    }
+
+    public Date getLastCompletedHour(long time) {


@mooxter These utility methods should be declared private as you only use it in this class. Or maybe create another class (AssignmentStatsUtil.java?) and move these methods there?

iizrailevsky · 2017-08-11T22:14:24Z

...tory-datastax/src/main/java/com/intuit/wasabi/repository/cassandra/impl/AssignmentStats.java

+    }
+
+    public UUID getExpUUID(Map.Entry pair){
+        String expIDString = pair.getKey().toString().substring(0, UUID_LENGTH);


@mooxter You might need to add error checking here to make sure the pair param is not null?

iizrailevsky · 2017-08-11T22:14:41Z

...tory-datastax/src/main/java/com/intuit/wasabi/repository/cassandra/impl/AssignmentStats.java

+    }
+
+    public String getBucketLabel(Map.Entry pair){
+        return pair.getKey().toString().substring(UUID_LENGTH);


@mooxter Same here - error checking...

iizrailevsky · 2017-08-11T22:15:54Z

...repository-datastax/src/main/java/com/intuit/wasabi/repository/cassandra/impl/ExpBucket.java

+
+
+class ExpBucket {
+    private Experiment.ID expID;


@mooxter These should probably be final as well, as they shouldn't change, right?

iizrailevsky · 2017-08-11T22:16:09Z

...repository-datastax/src/main/java/com/intuit/wasabi/repository/cassandra/impl/ExpBucket.java

+    }
+
+    public static String getKey(Experiment.ID expID, Bucket.Label bucket){
+        // TODO: Instead of concatenating two strings, redefine hashcode and equals methods.


@mooxter 👍 on the TODO

iizrailevsky · 2017-08-11T22:18:43Z

...-datastax/src/test/java/com/intuit/wasabi/repository/cassandra/impl/AssignmentStatsTest.java

+        Experiment experiment = mock(Experiment.class);
+        Assignment assignment = mock(Assignment.class);
+
+        String dateInString = "2016-11-08 16";


@mooxter Mind adding a few inline comments to describe what you're testing?

@iizrailevsky done.

iizrailevsky

@mooxter Looks good overall! Some comments, questions, and requests. Once resolved and TODO's are followed up on, & tests pass -> 👍

maxBruckhaus · 2017-08-15T20:57:47Z

...datastax/src/main/java/com/intuit/wasabi/repository/cassandra/CassandraRepositoryModule.java

+        //Bind Scheduled Executor Service
+        bind(ScheduledExecutorService.class).annotatedWith(named(ASSIGNMENTS_HOURLY_AGGREGATOR_SERVICE)).to(AssignmentCountScheduledExecutorService.class).in(Singleton.class);
+        //This is the assignment aggregation interval.
+        Integer assignmentAggregationIntervalInMinutes = 2;


This will be changed to 60 minutes for hourly aggregation, but it is 2 at the moment for testing purposes.

bargenilesh · 2017-08-16T21:32:19Z

...datastax/src/main/java/com/intuit/wasabi/repository/cassandra/CassandraRepositoryModule.java

 import static com.google.inject.name.Names.named;
 import static com.intuit.autumn.utils.PropertyFactory.create;
 import static com.intuit.autumn.utils.PropertyFactory.getProperty;
 import static java.lang.Integer.parseInt;
 import static org.slf4j.LoggerFactory.getLogger;

 public class CassandraRepositoryModule extends AbstractModule {
+    public String ASSIGNMENTS_HOURLY_AGGREGATOR_SERVICE = "AssignmentsHourlyAggregatorService";
+    public String ASSIGNMENTS_AGGREGATOR_INTERVAL = "AssignmentsAggregatorInterval";
+    public String ASSIGNMENTS_HOURLY_AGGREGATOR_TASK = "AssignmentHourlyAggregatorTask";


@mooxter it would be great to define all the related constants in AssignmentStatsAnnotations...

Turns out I had the constants defined in AssignmentStatsAnnotations and in CassandraRepositoryModule. Thanks for catching this, I just fixed it 👍

bargenilesh · 2017-08-16T21:34:42Z

...in/java/com/intuit/wasabi/repository/cassandra/accessor/count/HourlyBucketCountAccessor.java

+public interface HourlyBucketCountAccessor {
+    @Query("UPDATE hourly_bucket_counts SET hourly_bucket_count = hourly_bucket_count + 1 " +
+            "WHERE experiment_id = ? and day = ? and bucket_label = ? and event_time_hour = ? ")
+    ResultSet incrementCountBy(UUID experimentId, String day, String bucketLabel, int eventTimeHour);


@mooxter Major: it would be great to increment counter by given 'N' number rather than just 1

bargenilesh · 2017-08-16T21:37:12Z

...ava/com/intuit/wasabi/repository/cassandra/impl/AssignmentCountScheduledExecutorService.java

+
+    @Override
+    public void run() {
+        assignmentStats.writeCounts();


@mooxter Major: In order to make this scheduled executor service robust, you need to wrap complete run() method implementation in TRY-CATCH block, so that in case of any exception scheduled thread will be LIVE

@mooxter Also, if you add a INFO log here when this executor service starts & finishes with time taken to track the working of this service that would be great... similar to ScheduledCacheRefreshService...

bargenilesh · 2017-08-16T21:42:00Z

...tory-datastax/src/main/java/com/intuit/wasabi/repository/cassandra/impl/AssignmentStats.java

+        if (oldCount == null) {
+            synchronized (lock) {                         // oldCount would be initialized to 1 twice w/o this
+                oldCount = hourMap.get(key);
+                if (oldCount == null) {                   // double-checked locking


@mooxter 👍 for double-checked locking

bargenilesh · 2017-08-16T21:43:17Z

...tory-datastax/src/main/java/com/intuit/wasabi/repository/cassandra/impl/AssignmentStats.java

+     * @param experiment
+     * @param assignment
+     */
+    public void incrementCount(Experiment experiment, Assignment assignment) {


@mooxter it would be great to add execution flow debug logging with some inline comments of what is being done...

bargenilesh · 2017-08-16T22:03:04Z

...tory-datastax/src/main/java/com/intuit/wasabi/repository/cassandra/impl/AssignmentStats.java

+        String day = AssignmentStatsUtil.getDayString(completedHour);
+
+        for (String key : hourlyCountMap.get(assignmentHour).keySet()){
+            String experimentID = key.substring(0, UUID_LENGTH);


@mooxter it would be better to use 'ExpBucket' as a key for second level map so that you dont have to deal with the parsing logic here... E.g. Map<Integer, Map<ExpBucket, AtomicInteger>> hourlyCountMap - instead of - Map<Integer, Map<String, AtomicInteger>> hourlyCountMap

bargenilesh · 2017-08-16T22:06:04Z

...-datastax/src/main/java/com/intuit/wasabi/repository/cassandra/impl/AssignmentStatsUtil.java

+     * @param date the date containing the last completed hour
+     * @return int representing an hour of the day
+     */
+    static int getHour(Date date) {


@mooxter Major: SimpleDateFormatter is not thread-safe so here thread might run into race condition and so data corruption...

I switched usage of SimpleDateFormatter to DateTimeFormatter. I also replaced usage of Date with DateTime. I used the parsers from Joda Time instead, as they are thread safe.

bargenilesh · 2017-08-16T22:07:20Z

...repository-datastax/src/main/java/com/intuit/wasabi/repository/cassandra/impl/ExpBucket.java

+import com.intuit.wasabi.experimentobjects.Bucket;
+
+
+class ExpBucket {


@mooxter Minor: This class can be renamed better may be ExpBucket-> ExperimentBucketKey

@mooxter Major: This class needs to overrides equals() & hashcode() methods, probably add Builder as well...

bargenilesh · 2017-08-16T22:08:37Z

...repository-datastax/src/main/java/com/intuit/wasabi/repository/cassandra/impl/ExpBucket.java

@@ -0,0 +1,23 @@
+package com.intuit.wasabi.repository.cassandra.impl;


@mooxter Needs copyrights statement here...

bargenilesh · 2017-08-16T22:16:11Z

modules/repository-datastax/src/main/resources/repository.properties

@@ -19,3 +19,4 @@ export.pool.size:5
 default.time.format:${default.time.format}
 cassandra.mutagen.root.resource.path:${cassandra.mutagen.root.resource.path}
 mysql.mutagen.root.resource.path:${mysql.mutagen.root.resource.path}
+bucket.count.aggregation.interval:60


@mooxter here value for bucket.count.aggregation.interval can be taken from pom.xml - similar to other configuration keys...

Bruckhaus added 27 commits July 27, 2017 23:41

Add two nested concurrent hashmaps to keep track of bucket counts

01b8e1b

Add migration script and accessor

a80a850

Add HourlyBucketCount.java

c5dbe5e

Edit updateBucketAssignmentCount method to populate maps with bucket …

249d861

…counts

Add AssignmentStats and AssignmentStatsTest

ca092a1

Fix fillMaps method and fillMaps test

8b1c3be

Rename fillMaps to populateMaps

f4fbd8b

Change populateMaps to incrementCount, change map design, add unit te…

27f5eff

…sts, cleanup code

Update incrementCount method, add constructor, and cleanup code

b9af026

Merge branch 'develop' of https://github.com/intuit/wasabi into featu…

f423d24

…re/hourly-counts

Test improvements and updates

be255f5

Made AssignmentStats public, add scheduled executor service

c0f5738

Name change to AssignmentsHourlyAggregatorTask

ba34393

Name change to AssignmentsHourlyAggregatorTask

ab5aad2

Update hourly aggregator scheduled executor service

0b7cdab

Bind aggregator task

e807220

Add day String to incrementCount for hourlyBucketCounts

23d6230

Add code to write hourly counts to database using scheduled executor …

57c88db

…service

Merge branch 'develop' of https://github.com/intuit/wasabi into featu…

bb6efaa

…re/hourly-counts

Update writeCountst / code cleanup

c8b6beb

Update write counts

971ede2

Optimize imports

4af1e1e

Merge branch 'develop' of https://github.com/intuit/wasabi into featu…

7ca0053

…re/hourly-counts

Update imports

9cc4f1a

Update imports

97fd82c

Small change

965874c

Fix unit test

66996b4

maxBruckhaus requested review from abhishekjain88, sudheerkethireddy and bargenilesh August 10, 2017 19:06

maxBruckhaus requested a review from iizrailevsky August 10, 2017 19:06

Small cleanup

4640fc7

maxBruckhaus commented Aug 10, 2017

View reviewed changes

iizrailevsky reviewed Aug 11, 2017

View reviewed changes

iizrailevsky approved these changes Aug 11, 2017

View reviewed changes

Bruckhaus added 2 commits August 13, 2017 21:27

Addressing pull request comments

dc8d0d2

Fix Guice issues

decef0e

maxBruckhaus commented Aug 15, 2017

View reviewed changes

Bruckhaus added 3 commits August 15, 2017 23:38

Fix scheduled executor service to write hourly counts to db

6847a8a

Simplify writeCounts method

856660e

Improve incrementCount / incrementCountTest

15a9006

bargenilesh requested changes Aug 16, 2017

View reviewed changes

Bruckhaus added 5 commits August 16, 2017 20:25

Addressing pull request comments

f4b506e

Resolve TODO: Using ExperimentBucketKey as a key for second level map

280b2be

Replaced SimpleDateFormatter usage with thread safe DateTimeFormatter

2114ee7

Simplify date formatting logic

c32202d

Simplify getLastCompletedHour method

fc510fa

		import com.intuit.wasabi.experimentobjects.Bucket;


		class ExpBucket {

		@@ -0,0 +1,23 @@
		package com.intuit.wasabi.repository.cassandra.impl;

Feature/hourly counts #274

Are you sure you want to change the base?

Feature/hourly counts #274

Conversation

maxBruckhaus commented Aug 10, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

iizrailevsky Aug 11, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

iizrailevsky Aug 11, 2017 • edited

Choose a reason for hiding this comment

maxBruckhaus Aug 11, 2017 • edited

Choose a reason for hiding this comment

iizrailevsky Aug 11, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

maxBruckhaus Aug 11, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

iizrailevsky left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bargenilesh Aug 16, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bargenilesh Aug 16, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bargenilesh Aug 16, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bargenilesh Aug 16, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

maxBruckhaus commented Aug 10, 2017 •

edited

iizrailevsky Aug 11, 2017 •

edited

iizrailevsky Aug 11, 2017 •

edited

maxBruckhaus Aug 11, 2017 •

edited

iizrailevsky Aug 11, 2017 •

edited

maxBruckhaus Aug 11, 2017 •

edited

bargenilesh Aug 16, 2017 •

edited

bargenilesh Aug 16, 2017 •

edited

bargenilesh Aug 16, 2017 •

edited

bargenilesh Aug 16, 2017 •

edited