Add support for Alerta.io notifications (linkedin#1510)

.gitignore

@@ -16,3 +16,4 @@ logs
 target/
 access.log
 *.egg
+/bin/

cruise-control-core/src/main/java/com/linkedin/cruisecontrol/CruiseControlUtils.java

@@ -5,11 +5,13 @@
 package com.linkedin.cruisecontrol;
 
 import java.time.Instant;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeFormatterBuilder;
 import java.time.temporal.ChronoUnit;
+import java.time.temporal.TemporalUnit;
 
 import static com.linkedin.cruisecontrol.common.utils.Utils.validateNotNull;
 
-
 /**
  * Utils class for Cruise Control
  */
@@ -32,7 +34,7 @@ public static void ensureValidString(String fieldName, String toCheck) {
    * @return The current UTC date.
    */
   public static String currentUtcDate() {
-    return Instant.now().truncatedTo(ChronoUnit.SECONDS).toString();
+    return utcDateFor(Instant.now().getEpochSecond() * 1000);
   }
 
   /**
@@ -41,6 +43,20 @@ public static String currentUtcDate() {
    * @return The date for the given time in ISO 8601 format with date, hour, minute, and seconds.
    */
   public static String utcDateFor(long timeMs) {
-    return Instant.ofEpochMilli(timeMs).truncatedTo(ChronoUnit.SECONDS).toString();
+    return utcDateFor(timeMs, 0, ChronoUnit.SECONDS);
   }
+
+  /**
+   * @see <a href="https://xkcd.com/1179/">https://xkcd.com/1179/</a>
+   * @param timeMs Time in milliseconds.
+   * @param precision requested time precision used in {@link DateTimeFormatterBuilder#appendInstant()}
+   *        i.e: 0 for seconds precision, 3 for milliseconds, 6 for microseconds etc...
+   * @param roundTo round the provided time to the provided {@link TemporalUnit}
+   * @return The date for the given time in ISO 8601 format with provided precision (not truncated even if 0)
+   */
+  public static String utcDateFor(long timeMs, int precision, TemporalUnit roundTo) {
+    DateTimeFormatter formatter = new DateTimeFormatterBuilder().appendInstant(precision).toFormatter();
+    return formatter.format(Instant.ofEpochMilli(timeMs).truncatedTo(roundTo));
+  }
+
 }

cruise-control-core/src/test/java/com/linkedin/cruisecontrol/CruiseControlUtilsTest.java

@@ -37,6 +37,7 @@ public void testUtcDate() {
     long difference = currentUtcDate.minus(currentUtcDateFor.getEpochSecond(), ChronoUnit.SECONDS).getEpochSecond();
 
     // Maximum difference accounts for a potential GC pause -- ideally (1) currentUtcDateFor and (2) currentUtcDate should be the same.
-    assertTrue(difference >= 0 && difference < MAX_ESTIMATED_PAUSE);
+    assertTrue(difference >= 0);
+    assertTrue(difference < MAX_ESTIMATED_PAUSE);
   }
 }

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/TopicPartitionSizeAnomaly.java

@@ -4,6 +4,7 @@
 
 package com.linkedin.kafka.cruisecontrol.detector;
 
+import java.util.Collections;
 import java.util.Map;
 import java.util.function.Supplier;
 import org.apache.kafka.common.TopicPartition;
@@ -23,6 +24,13 @@
 public class TopicPartitionSizeAnomaly extends TopicAnomaly {
   protected Map<TopicPartition, Double> _sizeByPartition;
 
+  /**
+   * @return An unmodifiable version of the actual bad topic partitions size
+   */
+  public Map<TopicPartition, Double> getSizeByPartition() {
+    return Collections.unmodifiableMap(_sizeByPartition);
+  }
+
   /**
    * Fix the anomaly.
    *

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/TopicReplicationFactorAnomaly.java

@@ -8,6 +8,7 @@
 import com.linkedin.kafka.cruisecontrol.config.KafkaCruiseControlConfig;
 import com.linkedin.kafka.cruisecontrol.detector.notifier.KafkaAnomalyType;
 import com.linkedin.kafka.cruisecontrol.servlet.handler.async.runnable.UpdateTopicConfigurationRunnable;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
@@ -34,6 +35,13 @@ public class TopicReplicationFactorAnomaly extends TopicAnomaly {
   protected Map<Short, Set<TopicReplicationFactorAnomalyEntry>> _badTopicsByReplicationFactor;
   protected UpdateTopicConfigurationRunnable _updateTopicConfigurationRunnable;
 
+  /**
+   * @return An unmodifiable version of the actual bad topic map
+   */
+  public Map<Short, Set<TopicReplicationFactorAnomalyEntry>> getBadTopicsByReplicationFactor() {
+    return Collections.unmodifiableMap(_badTopicsByReplicationFactor);
+  }
+
   @Override
   public boolean fix() throws Exception {
     if (_updateTopicConfigurationRunnable == null) {

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/notifier/AlertSeverity.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2021 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License"). See License in the project root for license information.
+ */
+
+package com.linkedin.kafka.cruisecontrol.detector.notifier;
+
+/**
+ * Alert severity representation based on standard wording 
+ */
+public enum AlertSeverity {
+    CRITICAL("critical"), MAJOR("major"), MINOR("minor"), WARNING("warning");
+
+    private final String _value;
+
+    AlertSeverity(String value) {
+        this._value = value;
+    }
+
+    @Override
+    public String toString() {
+        return _value;
+    }
+}

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/notifier/AlertaAlertGroup.java

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2021 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License"). See License in the project root for license information.
+ */
+
+package com.linkedin.kafka.cruisecontrol.detector.notifier;
+
+/**
+ * @see <a href="https://docs.alerta.io/en/latest/conventions.html#event-groups">the Alerta event groups conventions</a>
+ */
+public enum AlertaAlertGroup {
+    PERFORMANCE("Performance"), STORAGE("Storage");
+
+    private final String _value;
+
+    AlertaAlertGroup(String value) {
+        this._value = value;
+    }
+
+    @Override
+    public String toString() {
+        return _value;
+    }
+}

cruise-control/src/main/java/com/linkedin/kafka/cruisecontrol/detector/notifier/AlertaMessage.java

@@ -0,0 +1,214 @@
+/*
+ * Copyright 2021 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License"). See License in the project root for license information.
+ */
+
+package com.linkedin.kafka.cruisecontrol.detector.notifier;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+import javax.validation.constraints.NotNull;
+import com.fasterxml.jackson.annotation.JsonInclude.Include;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.MapperFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+
+/**
+ * Alerta.io data to create alert: https://docs.alerta.io/en/latest/api/reference.html#create-an-alert
+ */
+public final class AlertaMessage implements Serializable {
+    private static final long serialVersionUID = -7290861136323903837L;
+
+    @NotNull
+    @JsonProperty("resource")
+    private String _resource;
+    @NotNull
+    @JsonProperty("event")
+    private String _event;
+
+    @JsonProperty("environment")
+    private String _environment;
+    @JsonProperty("severity")
+    private String _severity;
+    @JsonProperty("correlate")
+    private List<String> _correlate;
+    @JsonProperty("status")
+    private List<String> _status;
+    @JsonProperty("service")
+    private List<String> _service;
+    @JsonProperty("group")
+    private String _group;
+    @JsonProperty("value")
+    private String _value;
+    @JsonProperty("text")
+    private String _text;
+    @JsonProperty("tags")
+    private List<String> _tags;
+    @JsonProperty("attributes")
+    private Map<String, String> _attributes;
+    @JsonProperty("origin")
+    private String _origin;
+    @JsonProperty("type")
+    private String _type;
+    @JsonProperty("createTime")
+    private String _createTime;
+    @JsonProperty("timeout")
+    private String _timeout;
+    @JsonProperty("rawData")
+    private String _rawData;
+
+    public AlertaMessage(String resource, String event) {
+        this._event = event;
+        this._resource = resource;
+    }
+
+    public String getResource() {
+        return _resource;
+    }
+
+    public void setResource(String resource) {
+        this._resource = resource;
+    }
+
+    public String getEvent() {
+        return _event;
+    }
+
+    public void setEvent(String event) {
+        this._event = event;
+    }
+
+    public String getEnvironment() {
+        return _environment;
+    }
+
+    public void setEnvironment(String environment) {
+        this._environment = environment;
+    }
+
+    public String getSeverity() {
+        return _severity;
+    }
+
+    public void setSeverity(String severity) {
+        this._severity = severity;
+    }
+
+    public List<String> getCorrelate() {
+        return _correlate;
+    }
+
+    public void setCorrelate(List<String> correlate) {
+        this._correlate = correlate;
+    }
+
+    public List<String> getService() {
+        return _service;
+    }
+
+    public void setService(List<String> service) {
+        this._service = service;
+    }
+
+    public String getGroup() {
+        return _group;
+    }
+
+    public void setGroup(String group) {
+        this._group = group;
+    }
+
+    public String getValue() {
+        return _value;
+    }
+
+    public void setValue(String value) {
+        this._value = value;
+    }
+
+    public String getText() {
+        return _text;
+    }
+
+    public void setText(String text) {
+        this._text = text;
+    }
+
+    public List<String> getTags() {
+        return _tags;
+    }
+
+    public void setTags(List<String> tags) {
+        this._tags = tags;
+    }
+
+    public String getOrigin() {
+        return _origin;
+    }
+
+    public void setOrigin(String origin) {
+        this._origin = origin;
+    }
+
+    public String getType() {
+        return _type;
+    }
+
+    public void setType(String type) {
+        this._type = type;
+    }
+
+    public String getCreateTime() {
+        return _createTime;
+    }
+
+    public void setCreateTime(String createTime) {
+        this._createTime = createTime;
+    }
+
+    public String getTimeout() {
+        return _timeout;
+    }
+
+    public void setTimeout(String timeout) {
+        this._timeout = timeout;
+    }
+
+    public String getRawData() {
+        return _rawData;
+    }
+
+    public void setRawData(String rawData) {
+        this._rawData = rawData;
+    }
+
+    public List<String> getStatus() {
+        return _status;
+    }
+
+    public void setStatus(List<String> status) {
+        this._status = status;
+    }
+
+    public Map<String, String> getAttributes() {
+        return _attributes;
+    }
+
+    public void setAttributes(Map<String, String> attributes) {
+        this._attributes = attributes;
+    }
+
+    @Override
+    public String toString() {
+        try {
+            return new ObjectMapper().configure(MapperFeature.SORT_PROPERTIES_ALPHABETICALLY, true)
+                .configure(SerializationFeature.ORDER_MAP_ENTRIES_BY_KEYS, true)
+                .setSerializationInclusion(Include.NON_NULL).writeValueAsString(this);
+        } catch (JsonProcessingException e) {
+            return "AlertaMassage Object parsing error : " + e.getMessage();
+        }
+    }
+}