Core: Migrating from joda to java.time. Monitoring plugin (#36297)

monitoring plugin migration from joda to java.time

refers #27330

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/monitoring/exporter/MonitoringDoc.java

@@ -10,20 +10,23 @@
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.common.xcontent.ToXContentObject;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.xpack.core.monitoring.MonitoredSystem;
-import org.joda.time.DateTime;
-import org.joda.time.DateTimeZone;
 
 import java.io.IOException;
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
 import java.util.Objects;
 
 /**
  * Base class for all monitoring documents.
  */
 public abstract class MonitoringDoc implements ToXContentObject {
 
+    private static final DateFormatter dateTimeFormatter = DateFormatter.forPattern("strict_date_time");
     private final String cluster;
     private final long timestamp;
     private final long intervalMillis;
@@ -123,7 +126,9 @@ public XContentBuilder toXContent(XContentBuilder builder, Params params) throws
      * @return a string representing the timestamp
      */
     public static String toUTC(final long timestamp) {
-        return new DateTime(timestamp, DateTimeZone.UTC).toString();
+        ZonedDateTime zonedDateTime = Instant.ofEpochMilli(timestamp).atZone(ZoneOffset.UTC);
+        return dateTimeFormatter.format(zonedDateTime);
+
     }
 
     /**
@@ -250,4 +255,4 @@ public int hashCode() {
             return Objects.hash(uuid, host, transportAddress, ip, name, timestamp);
         }
     }
-}
+}

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/monitoring/exporter/MonitoringTemplateUtils.java

@@ -6,15 +6,16 @@
 package org.elasticsearch.xpack.core.monitoring.exporter;
 
 import org.elasticsearch.Version;
+import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.xpack.core.monitoring.MonitoredSystem;
 import org.elasticsearch.xpack.core.template.TemplateUtils;
-import org.joda.time.format.DateTimeFormatter;
-import org.elasticsearch.common.Strings;
 
 import java.io.IOException;
+import java.time.Instant;
 import java.util.Locale;
 import java.util.regex.Pattern;
 
@@ -252,12 +253,12 @@ public static XContentBuilder emptyPipeline(final XContentType type) {
     /**
      * Get the index name given a specific date format, a monitored system and a timestamp.
      *
-     * @param formatter the {@link DateTimeFormatter} to use to compute the timestamped index name
+     * @param formatter the {@link DateFormatter} to use to compute the timestamped index name
      * @param system the {@link MonitoredSystem} for which the index name is computed
      * @param timestamp the timestamp value to use to compute the timestamped index name
      * @return the index name as a @{link String}
      */
-    public static String indexName(final DateTimeFormatter formatter, final MonitoredSystem system, final long timestamp) {
-        return ".monitoring-" + system.getSystem() + "-" + TEMPLATE_VERSION + "-" + formatter.print(timestamp);
+    public static String indexName(final DateFormatter formatter, final MonitoredSystem system, final long timestamp) {
+        return ".monitoring-" + system.getSystem() + "-" + TEMPLATE_VERSION + "-" + formatter.format(Instant.ofEpochMilli(timestamp));
     }
 }

x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/cleaner/CleanerService.java

@@ -17,9 +17,10 @@
 import org.elasticsearch.threadpool.Scheduler;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.xpack.core.monitoring.MonitoringField;
-import org.joda.time.DateTime;
-import org.joda.time.chrono.ISOChronology;
 
+import java.time.Clock;
+import java.time.Duration;
+import java.time.ZonedDateTime;
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 
@@ -56,7 +57,8 @@ public CleanerService(Settings settings, ClusterSettings clusterSettings, Thread
     @Override
     protected void doStart() {
         logger.debug("starting cleaning service");
-        threadPool.schedule(runnable, executionScheduler.nextExecutionDelay(new DateTime(ISOChronology.getInstance())), executorName());
+        threadPool.schedule(runnable, executionScheduler.nextExecutionDelay(ZonedDateTime.now(Clock.systemDefaultZone())),
+            executorName());
         logger.debug("cleaning service started");
     }
 
@@ -190,7 +192,7 @@ protected void doRunInLifecycle() throws Exception {
          */
         @Override
         protected void onAfterInLifecycle() {
-            DateTime start = new DateTime(ISOChronology.getInstance());
+            ZonedDateTime start = ZonedDateTime.now(Clock.systemUTC());
             TimeValue delay = executionScheduler.nextExecutionDelay(start);
 
             logger.debug("scheduling next execution in [{}] seconds", delay.seconds());
@@ -233,7 +235,7 @@ interface ExecutionScheduler {
          * @param now the current time
          * @return the delay in millis
          */
-        TimeValue nextExecutionDelay(DateTime now);
+        TimeValue nextExecutionDelay(ZonedDateTime now);
     }
 
     /**
@@ -242,14 +244,16 @@ interface ExecutionScheduler {
     static class DefaultExecutionScheduler implements ExecutionScheduler {
 
         @Override
-        public TimeValue nextExecutionDelay(DateTime now) {
+        public TimeValue nextExecutionDelay(ZonedDateTime now) {
             // Runs at 01:00 AM today or the next day if it's too late
-            DateTime next = now.withTimeAtStartOfDay().plusHours(1);
+            ZonedDateTime next = now.toLocalDate()
+                .atStartOfDay(now.getZone())
+                .plusHours(1);
             // if it's not after now, then it needs to be the next day!
             if (next.isAfter(now) == false) {
                 next = next.plusDays(1);
             }
-            return TimeValue.timeValueMillis(next.getMillis() - now.getMillis());
+            return TimeValue.timeValueMillis(Duration.between(now, next).toMillis());
         }
     }
 }

x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/exporter/Exporter.java

@@ -11,10 +11,10 @@
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsException;
+import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.license.XPackLicenseState;
-import org.joda.time.format.DateTimeFormat;
-import org.joda.time.format.DateTimeFormatter;
 
+import java.time.ZoneOffset;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
@@ -70,7 +70,7 @@ public abstract class Exporter implements AutoCloseable {
             Setting.affixKeySetting("xpack.monitoring.exporters.","index.name.time_format",
                     key -> Setting.simpleString(key, Property.Dynamic, Property.NodeScope));
 
-    private static final String INDEX_FORMAT = "YYYY.MM.dd";
+    private static final String INDEX_FORMAT = "yyyy.MM.dd";
 
     protected final Config config;
 
@@ -113,11 +113,11 @@ public void close() {
 
     protected abstract void doClose();
 
-    protected static DateTimeFormatter dateTimeFormatter(final Config config) {
+    protected static DateFormatter dateTimeFormatter(final Config config) {
         Setting<String> setting = INDEX_NAME_TIME_FORMAT_SETTING.getConcreteSettingForNamespace(config.name);
         String format = setting.exists(config.settings()) ? setting.get(config.settings()) : INDEX_FORMAT;
         try {
-            return DateTimeFormat.forPattern(format).withZoneUTC();
+            return DateFormatter.forPattern(format).withZone(ZoneOffset.UTC);
         } catch (IllegalArgumentException e) {
             throw new SettingsException("[" + INDEX_NAME_TIME_FORMAT_SETTING.getKey() + "] invalid index name time format: ["
                     + format + "]", e);

x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/exporter/http/HttpExportBulk.java

@@ -19,6 +19,7 @@
 import org.elasticsearch.client.RestClient;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.io.stream.BytesStreamOutput;
+import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.xcontent.XContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
@@ -28,9 +29,9 @@
 import org.elasticsearch.xpack.core.monitoring.exporter.MonitoringTemplateUtils;
 import org.elasticsearch.xpack.monitoring.exporter.ExportBulk;
 import org.elasticsearch.xpack.monitoring.exporter.ExportException;
-import org.joda.time.format.DateTimeFormatter;
 
 import java.io.IOException;
+import java.time.format.DateTimeFormatter;
 import java.util.Collection;
 import java.util.Map;
 
@@ -54,15 +55,15 @@ class HttpExportBulk extends ExportBulk {
     /**
      * {@link DateTimeFormatter} used to resolve timestamped index name.
      */
-    private final DateTimeFormatter formatter;
+    private final DateFormatter formatter;
 
     /**
      * The bytes payload that represents the bulk body is created via {@link #doAdd(Collection)}.
      */
     private byte[] payload = null;
 
     HttpExportBulk(final String name, final RestClient client, final Map<String, String> parameters,
-                   final DateTimeFormatter dateTimeFormatter, final ThreadContext threadContext) {
+                   final DateFormatter dateTimeFormatter, final ThreadContext threadContext) {
         super(name, threadContext);
 
         this.client = client;

x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/exporter/http/HttpExporter.java

@@ -30,6 +30,7 @@
 import org.elasticsearch.common.settings.Setting.Property;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.settings.SettingsException;
+import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.util.set.Sets;
@@ -41,7 +42,6 @@
 import org.elasticsearch.xpack.monitoring.exporter.ClusterAlertsUtil;
 import org.elasticsearch.xpack.monitoring.exporter.ExportBulk;
 import org.elasticsearch.xpack.monitoring.exporter.Exporter;
-import org.joda.time.format.DateTimeFormatter;
 
 import javax.net.ssl.SSLContext;
 import java.util.ArrayList;
@@ -193,7 +193,7 @@ public class HttpExporter extends Exporter {
     private final AtomicBoolean clusterAlertsAllowed = new AtomicBoolean(false);
 
     private final ThreadContext threadContext;
-    private final DateTimeFormatter dateTimeFormatter;
+    private final DateFormatter dateTimeFormatter;
 
     /**
      * Create an {@link HttpExporter}.

x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/exporter/local/LocalBulk.java

@@ -14,13 +14,13 @@
 import org.elasticsearch.client.Client;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.xpack.core.monitoring.exporter.MonitoringDoc;
 import org.elasticsearch.xpack.core.monitoring.exporter.MonitoringTemplateUtils;
 import org.elasticsearch.xpack.monitoring.exporter.ExportBulk;
 import org.elasticsearch.xpack.monitoring.exporter.ExportException;
-import org.joda.time.format.DateTimeFormatter;
 
 import java.util.Arrays;
 import java.util.Collection;
@@ -37,13 +37,13 @@ public class LocalBulk extends ExportBulk {
 
     private final Logger logger;
     private final Client client;
-    private final DateTimeFormatter formatter;
+    private final DateFormatter formatter;
     private final boolean usePipeline;
 
     private BulkRequestBuilder requestBuilder;
 
 
-    LocalBulk(String name, Logger logger, Client client, DateTimeFormatter dateTimeFormatter, boolean usePipeline) {
+    LocalBulk(String name, Logger logger, Client client, DateFormatter dateTimeFormatter, boolean usePipeline) {
         super(name, client.threadPool().getThreadContext());
         this.logger = logger;
         this.client = client;

x-pack/plugin/monitoring/src/main/java/org/elasticsearch/xpack/monitoring/exporter/local/LocalExporter.java

@@ -28,6 +28,7 @@
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.regex.Regex;
+import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.xcontent.XContentType;
@@ -51,10 +52,11 @@
 import org.elasticsearch.xpack.monitoring.exporter.ClusterAlertsUtil;
 import org.elasticsearch.xpack.monitoring.exporter.ExportBulk;
 import org.elasticsearch.xpack.monitoring.exporter.Exporter;
-import org.joda.time.DateTime;
-import org.joda.time.DateTimeZone;
-import org.joda.time.format.DateTimeFormatter;
 
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.time.temporal.ChronoUnit;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
@@ -89,7 +91,7 @@ public class LocalExporter extends Exporter implements ClusterStateListener, Cle
     private final XPackLicenseState licenseState;
     private final CleanerService cleanerService;
     private final boolean useIngest;
-    private final DateTimeFormatter dateTimeFormatter;
+    private final DateFormatter dateTimeFormatter;
     private final List<String> clusterAlertBlacklist;
 
     private final AtomicReference<State> state = new AtomicReference<>(State.INITIALIZED);
@@ -489,12 +491,12 @@ public void onCleanUpIndices(TimeValue retention) {
         if (clusterService.state().nodes().isLocalNodeElectedMaster()) {
             // Reference date time will be compared to index.creation_date settings,
             // that's why it must be in UTC
-            DateTime expiration = new DateTime(DateTimeZone.UTC).minus(retention.millis());
+            ZonedDateTime expiration = ZonedDateTime.now(ZoneOffset.UTC).minus(retention.millis(), ChronoUnit.MILLIS);
             logger.debug("cleaning indices [expiration={}, retention={}]", expiration, retention);
 
             ClusterState clusterState = clusterService.state();
             if (clusterState != null) {
-                final long expirationTimeMillis = expiration.getMillis();
+                final long expirationTimeMillis = expiration.toInstant().toEpochMilli();
                 final long currentTimeMillis = System.currentTimeMillis();
                 final boolean cleanUpWatcherHistory = clusterService.getClusterSettings().get(CLEAN_WATCHER_HISTORY);
 
@@ -524,7 +526,7 @@ public void onCleanUpIndices(TimeValue retention) {
                         if (creationDate <= expirationTimeMillis) {
                             if (logger.isDebugEnabled()) {
                                 logger.debug("detected expired index [name={}, created={}, expired={}]",
-                                        indexName, new DateTime(creationDate, DateTimeZone.UTC), expiration);
+                                        indexName, Instant.ofEpochMilli(creationDate).atZone(ZoneOffset.UTC), expiration);
                             }
                             indices.add(indexName);
                         }