Adding User support for Detector and DetectorJob

build.gradle

@@ -298,6 +298,8 @@ dependencies {
     compile "org.elasticsearch:elasticsearch:${es_version}"
     compileOnly "org.elasticsearch.plugin:elasticsearch-scripting-painless-spi:${versions.elasticsearch}"
     compileOnly "com.amazon.opendistroforelasticsearch:opendistro-job-scheduler-spi:1.10.1.1"
+    // Will be moved to Maven Depedency when https://github.com/opendistro-for-elasticsearch/common-utils repo publishes a release
+    compile files('libs/common-utils-1.10.1.0.jar')
     compile group: 'com.google.guava', name: 'guava', version:'15.0'
     compile group: 'org.apache.commons', name: 'commons-math3', version: '3.6.1'
     compile group: 'com.google.code.gson', name: 'gson', version: '2.8.5'
@@ -306,6 +308,7 @@ dependencies {
     compile group: 'commons-lang', name: 'commons-lang', version: '2.6'
     compile 'software.amazon.randomcutforest:randomcutforest-core:1.0'
     compile 'software.amazon.randomcutforest:randomcutforest-serialization-json:1.0'
+    compile "org.elasticsearch.client:elasticsearch-rest-client:${es_version}"
 
     compile "org.jacoco:org.jacoco.agent:0.8.5"
     compile ("org.jacoco:org.jacoco.ant:0.8.5") {

src/main/java/com/amazon/opendistroforelasticsearch/ad/AnomalyDetectorJobRunner.java

@@ -381,7 +381,8 @@ private void stopAdJob(String detectorId) {
                                 job.getEnabledTime(),
                                 Instant.now(),
                                 Instant.now(),
-                                job.getLockDurationSeconds()
+                                job.getLockDurationSeconds(),
+                                job.getUser()
                             );
                             IndexRequest indexRequest = new IndexRequest(AnomalyDetectorJob.ANOMALY_DETECTOR_JOB_INDEX)
                                 .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)

src/main/java/com/amazon/opendistroforelasticsearch/ad/model/AnomalyDetector.java

@@ -48,6 +48,7 @@
 
 import com.amazon.opendistroforelasticsearch.ad.annotation.Generated;
 import com.amazon.opendistroforelasticsearch.ad.util.ParseUtils;
+import com.amazon.opendistroforelasticsearch.commons.authuser.User;
 import com.google.common.base.Objects;
 
 /**
@@ -80,6 +81,7 @@ public class AnomalyDetector implements Writeable, ToXContentObject {
     private static final String LAST_UPDATE_TIME_FIELD = "last_update_time";
     public static final String UI_METADATA_FIELD = "ui_metadata";
     public static final String CATEGORY_FIELD = "category_field";
+    public static final String USER_FIELD = "user";
 
     private final String detectorId;
     private final Long version;
@@ -96,6 +98,7 @@ public class AnomalyDetector implements Writeable, ToXContentObject {
     private final Integer schemaVersion;
     private final Instant lastUpdateTime;
     private final List<String> categoryFields;
+    private User user;
 
     /**
      * Constructor function.
@@ -114,7 +117,8 @@ public class AnomalyDetector implements Writeable, ToXContentObject {
      * @param uiMetadata        metadata used by Kibana
      * @param schemaVersion     anomaly detector index mapping version
      * @param lastUpdateTime    detector's last update time
-     * @param categoryFields     a list of partition fields
+     * @param categoryFields    a list of partition fields
+     * @param user              user to which detector is associated
      */
     public AnomalyDetector(
         String detectorId,
@@ -131,7 +135,8 @@ public AnomalyDetector(
         Map<String, Object> uiMetadata,
         Integer schemaVersion,
         Instant lastUpdateTime,
-        List<String> categoryFields
+        List<String> categoryFields,
+        User user
     ) {
         if (Strings.isBlank(name)) {
             throw new IllegalArgumentException("Detector name should be set");
@@ -166,6 +171,7 @@ public AnomalyDetector(
         this.schemaVersion = schemaVersion;
         this.lastUpdateTime = lastUpdateTime;
         this.categoryFields = categoryFields;
+        this.user = user;
     }
 
     // TODO: remove after complete code merges. Created to not to touch too
@@ -184,7 +190,8 @@ public AnomalyDetector(
         Integer shingleSize,
         Map<String, Object> uiMetadata,
         Integer schemaVersion,
-        Instant lastUpdateTime
+        Instant lastUpdateTime,
+        User user
     ) {
         this(
             detectorId,
@@ -201,7 +208,8 @@ public AnomalyDetector(
             uiMetadata,
             schemaVersion,
             lastUpdateTime,
-            null
+            null,
+            user
         );
     }
 
@@ -237,6 +245,11 @@ public AnomalyDetector(StreamInput input) throws IOException {
         schemaVersion = input.readInt();
         lastUpdateTime = input.readInstant();
         this.categoryFields = input.readStringList();
+        if (input.readBoolean()) {
+            this.user = new User(input);
+        } else {
+            user = null;
+        }
     }
 
     public XContentBuilder toXContent(XContentBuilder builder) throws IOException {
@@ -260,6 +273,12 @@ public void writeTo(StreamOutput output) throws IOException {
         output.writeInt(schemaVersion);
         output.writeInstant(lastUpdateTime);
         output.writeStringCollection(categoryFields);
+        if (user != null) {
+            output.writeBoolean(true); // user exists
+            user.writeTo(output);
+        } else {
+            output.writeBoolean(false); // user does not exist
+        }
     }
 
     @Override
@@ -289,6 +308,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         if (categoryFields != null) {
             xContentBuilder.field(CATEGORY_FIELD, categoryFields.toArray());
         }
+        if (user != null) {
+            xContentBuilder.field(USER_FIELD, user);
+        }
         return xContentBuilder.endObject();
     }
 
@@ -354,6 +376,7 @@ public static AnomalyDetector parse(
         int schemaVersion = 0;
         Map<String, Object> uiMetadata = null;
         Instant lastUpdateTime = null;
+        User user = null;
 
         List<String> categoryField = null;
 
@@ -415,6 +438,9 @@ public static AnomalyDetector parse(
                 case CATEGORY_FIELD:
                     categoryField = (List) parser.list();
                     break;
+                case USER_FIELD:
+                    user = User.parse(parser);
+                    break;
                 default:
                     parser.skipChildren();
                     break;
@@ -435,7 +461,8 @@ public static AnomalyDetector parse(
             uiMetadata,
             schemaVersion,
             lastUpdateTime,
-            categoryField
+            categoryField,
+            user
         );
     }
 
@@ -583,4 +610,8 @@ public long getDetectorIntervalInSeconds() {
     public Duration getDetectionIntervalDuration() {
         return ((IntervalTimeConfiguration) getDetectionInterval()).toDuration();
     }
+
+    public User getUser() {
+        return user;
+    }
 }

src/main/java/com/amazon/opendistroforelasticsearch/ad/model/AnomalyDetectorJob.java

@@ -31,6 +31,7 @@
 import org.elasticsearch.common.xcontent.XContentParser;
 
 import com.amazon.opendistroforelasticsearch.ad.util.ParseUtils;
+import com.amazon.opendistroforelasticsearch.commons.authuser.User;
 import com.amazon.opendistroforelasticsearch.jobscheduler.spi.ScheduledJobParameter;
 import com.amazon.opendistroforelasticsearch.jobscheduler.spi.schedule.CronSchedule;
 import com.amazon.opendistroforelasticsearch.jobscheduler.spi.schedule.IntervalSchedule;
@@ -64,6 +65,7 @@ enum ScheduleType {
     private static final String IS_ENABLED_FIELD = "enabled";
     private static final String ENABLED_TIME_FIELD = "enabled_time";
     private static final String DISABLED_TIME_FIELD = "disabled_time";
+    public static final String USER_FIELD = "user";
 
     private final String name;
     private final Schedule schedule;
@@ -73,6 +75,7 @@ enum ScheduleType {
     private final Instant disabledTime;
     private final Instant lastUpdateTime;
     private final Long lockDurationSeconds;
+    private final User user;
 
     public AnomalyDetectorJob(
         String name,
@@ -82,7 +85,8 @@ public AnomalyDetectorJob(
         Instant enabledTime,
         Instant disabledTime,
         Instant lastUpdateTime,
-        Long lockDurationSeconds
+        Long lockDurationSeconds,
+        User user
     ) {
         this.name = name;
         this.schedule = schedule;
@@ -92,6 +96,7 @@ public AnomalyDetectorJob(
         this.disabledTime = disabledTime;
         this.lastUpdateTime = lastUpdateTime;
         this.lockDurationSeconds = lockDurationSeconds;
+        this.user = user;
     }
 
     public AnomalyDetectorJob(StreamInput input) throws IOException {
@@ -107,6 +112,11 @@ public AnomalyDetectorJob(StreamInput input) throws IOException {
         disabledTime = input.readInstant();
         lastUpdateTime = input.readInstant();
         lockDurationSeconds = input.readLong();
+        if (input.readBoolean()) {
+            user = new User(input);
+        } else {
+            user = null;
+        }
     }
 
     @Override
@@ -123,6 +133,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
         if (disabledTime != null) {
             xContentBuilder.field(DISABLED_TIME_FIELD, disabledTime.toEpochMilli());
         }
+        if (user != null) {
+            xContentBuilder.field(USER_FIELD, user);
+        }
         return xContentBuilder.endObject();
     }
 
@@ -140,6 +153,12 @@ public void writeTo(StreamOutput output) throws IOException {
         output.writeInstant(disabledTime);
         output.writeInstant(lastUpdateTime);
         output.writeLong(lockDurationSeconds);
+        if (user != null) {
+            output.writeBoolean(true); // user exists
+            user.writeTo(output);
+        } else {
+            output.writeBoolean(false); // user does not exist
+        }
     }
 
     public static AnomalyDetectorJob parse(XContentParser parser) throws IOException {
@@ -152,6 +171,7 @@ public static AnomalyDetectorJob parse(XContentParser parser) throws IOException
         Instant disabledTime = null;
         Instant lastUpdateTime = null;
         Long lockDurationSeconds = DEFAULT_AD_JOB_LOC_DURATION_SECONDS;
+        User user = null;
 
         ensureExpectedToken(XContentParser.Token.START_OBJECT, parser.currentToken(), parser::getTokenLocation);
         while (parser.nextToken() != XContentParser.Token.END_OBJECT) {
@@ -183,6 +203,9 @@ public static AnomalyDetectorJob parse(XContentParser parser) throws IOException
                 case LOCK_DURATION_SECONDS:
                     lockDurationSeconds = parser.longValue();
                     break;
+                case USER_FIELD:
+                    user = User.parse(parser);
+                    break;
                 default:
                     parser.skipChildren();
                     break;
@@ -196,7 +219,8 @@ public static AnomalyDetectorJob parse(XContentParser parser) throws IOException
             enabledTime,
             disabledTime,
             lastUpdateTime,
-            lockDurationSeconds
+            lockDurationSeconds,
+            user
         );
     }
 
@@ -258,4 +282,8 @@ public Instant getLastUpdateTime() {
     public Long getLockDurationSeconds() {
         return lockDurationSeconds;
     }
+
+    public User getUser() {
+        return user;
+    }
 }

src/main/java/com/amazon/opendistroforelasticsearch/ad/rest/handler/IndexAnomalyDetectorActionHandler.java

@@ -407,7 +407,8 @@ private void indexAnomalyDetector(String detectorId) throws IOException {
             anomalyDetector.getUiMetadata(),
             anomalyDetector.getSchemaVersion(),
             Instant.now(),
-            anomalyDetector.getCategoryField()
+            anomalyDetector.getCategoryField(),
+            anomalyDetector.getUser()
         );
         IndexRequest indexRequest = new IndexRequest(ANOMALY_DETECTORS_INDEX)
             .setRefreshPolicy(refreshPolicy)

src/main/java/com/amazon/opendistroforelasticsearch/ad/rest/handler/IndexAnomalyDetectorJobActionHandler.java

@@ -175,7 +175,8 @@ private void onGetAnomalyDetectorResponse(GetResponse response) throws IOExcepti
                 Instant.now(),
                 null,
                 Instant.now(),
-                duration.getSeconds()
+                duration.getSeconds(),
+                detector.getUser()
             );
 
             getAnomalyDetectorJobForWrite(job);
@@ -218,7 +219,8 @@ private void onGetAnomalyDetectorJobForWrite(GetResponse response, AnomalyDetect
                         Instant.now(),
                         currentAdJob.getDisabledTime(),
                         Instant.now(),
-                        job.getLockDurationSeconds()
+                        job.getLockDurationSeconds(),
+                        job.getUser()
                     );
                     indexAnomalyDetectorJob(newJob, null);
                 }
@@ -301,7 +303,8 @@ public void stopAnomalyDetectorJob(String detectorId) {
                             job.getEnabledTime(),
                             Instant.now(),
                             Instant.now(),
-                            job.getLockDurationSeconds()
+                            job.getLockDurationSeconds(),
+                            job.getUser()
                         );
                         indexAnomalyDetectorJob(
                             newJob,

src/test/java/com/amazon/opendistroforelasticsearch/ad/AnomalyDetectorJobRunnerTests.java

@@ -310,7 +310,8 @@ private void testRunAdJobWithEndRunExceptionNowAndStopAdJob(boolean jobExists, b
                                 Instant.now().minusSeconds(60),
                                 Instant.now(),
                                 Instant.now(),
-                                60L
+                                60L,
+                                TestHelpers.randomUser()
                             ).toXContent(TestHelpers.builder(), ToXContent.EMPTY_PARAMS)
                         ),
                     Collections.emptyMap(),

src/test/java/com/amazon/opendistroforelasticsearch/ad/AnomalyDetectorRestTestCase.java

@@ -107,7 +107,8 @@ protected AnomalyDetector createAnomalyDetector(AnomalyDetector detector, Boolea
             detector.getUiMetadata(),
             detector.getSchemaVersion(),
             detector.getLastUpdateTime(),
-            null
+            null,
+            detector.getUser()
         );
     }
 
@@ -178,7 +179,8 @@ public ToXContentObject[] getAnomalyDetector(String detectorId, BasicHeader head
                 detector.getUiMetadata(),
                 detector.getSchemaVersion(),
                 detector.getLastUpdateTime(),
-                null
+                null,
+                detector.getUser()
             ),
             detectorJob };
     }