SAMZA-1733 New metric type Listgauge

samza-api/src/main/java/org/apache/samza/metrics/DefaultListGaugeEvictionPolicy.java

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.metrics;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Queue;
+import org.apache.samza.util.TimestampedValue;
+
+
+/**
+ * Provides an eviction policy that evicts entries from the elements if
+ * a.) There are more elements in the elements than the specified maxNumberOfItems (removal in FIFO order), or
+ * b.) There are elements which have timestamps which are stale as compared to currentTime (the staleness bound is
+ * specified as maxStaleness).
+ *
+ */
+public class DefaultListGaugeEvictionPolicy<T> {
+
+  /**
+   * Evicts entries from the elements list, based on the given item-size and durationThreshold.
+   * Callers are responsible for thread-safety.
+   */
+  public void evict(Queue<TimestampedValue<T>> elements, int maxNumberOfItems, Duration maxStaleness) {
+    this.evictBasedOnSize(elements, maxNumberOfItems);
+    this.evictBasedOnTimestamp(elements, maxStaleness);
+  }
+
+  /**
+   * Evicts entries from elements in FIFO order until it has maxNumberOfItems
+   * @param elements queue to evict elements from
+   * @param maxNumberOfItems max number of items to be left in the queue
+   */
+  private void evictBasedOnSize(Queue<TimestampedValue<T>> elements, int maxNumberOfItems) {
+    int numToEvict = elements.size() - maxNumberOfItems;
+    while (numToEvict > 0) {
+      elements.poll(); // remove head
+      numToEvict--;
+    }
+  }
+
+  /**
+   * Removes entries from elements to ensure no element has a timestamp more than maxStaleness before current timestamp.
+   * @param elements the queue to evict elements from
+   * @param maxStaleness max staleness permitted in elements
+   */
+  private void evictBasedOnTimestamp(Queue<TimestampedValue<T>> elements, Duration maxStaleness) {
+    Instant currentTimestamp = Instant.now();
+    TimestampedValue<T> valueInfo = elements.peek();
+
+    // continue remove-head if currenttimestamp - head-element's timestamp > durationThreshold
+    while (valueInfo != null && currentTimestamp.toEpochMilli() - valueInfo.getTimestamp() > maxStaleness.toMillis()) {
+      elements.poll();
+      valueInfo = elements.peek();
+    }
+  }
+}

samza-api/src/main/java/org/apache/samza/metrics/ListGauge.java

@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.metrics;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.stream.Collectors;
+import org.apache.samza.util.TimestampedValue;
+
+
+/**
+ * A {@link ListGauge} is a {@link org.apache.samza.metrics.Metric} that buffers multiple instances of a type T in a list.
+ * {@link ListGauge}s are useful for maintaining, recording, or collecting values over time.
+ * For example, a set of specific logging-events (e.g., errors).
+ *
+ * Eviction from list is either done by consuming-code using the remove APIs or by specifying an eviction policy
+ * at creation time.
+ *
+ * All public methods are thread-safe.
+ *
+ */
+public class ListGauge<T> implements Metric {
+  private final String name;
+  private final Queue<TimestampedValue<T>> elements;
+  private final DefaultListGaugeEvictionPolicy listGaugeEvictionPolicy;
+
+  private final int maxNumberOfItems;
+  private final Duration maxStaleness;
+  private final static int DEFAULT_MAX_NITEMS = 1000;
+  private final static Duration DEFAULT_MAX_STALENESS = Duration.ofMinutes(60);
+
+  /**
+   * Create a new {@link ListGauge} that auto evicts based on the given maxNumberOfItems, maxStaleness, and period parameters.
+   *
+   * @param name Name to be assigned
+   * @param maxNumberOfItems The max number of items that can remain in the listgauge
+   * @param maxStaleness The max staleness of items permitted in the listgauge
+   */
+  public ListGauge(String name, int maxNumberOfItems, Duration maxStaleness) {
+    this.name = name;
+    this.elements = new ConcurrentLinkedQueue<TimestampedValue<T>>();
+    this.maxNumberOfItems = maxNumberOfItems;
+    this.maxStaleness = maxStaleness;
+    this.listGaugeEvictionPolicy = new DefaultListGaugeEvictionPolicy();
+  }
+
+  /**
+   * Create a new {@link ListGauge} that auto evicts upto a max of 100 items and a max-staleness of 60 minutes.
+   * @param name Name to be assigned
+   */
+  public ListGauge(String name) {
+    this(name, DEFAULT_MAX_NITEMS, DEFAULT_MAX_STALENESS);
+  }
+
+  /**
+   * Get the name assigned to this {@link ListGauge}
+   * @return the assigned name
+   */
+  public String getName() {
+    return this.name;
+  }
+
+  /**
+   * Get the Collection of Gauge values currently in the list, used when serializing this Gauge.
+   * Also evicts values based on the configured maxItems and maxStaleness.
+   * @return the collection of gauge values
+   */
+  public Collection<T> getValues() {
+    // notify the policy object for performing any eviction that may be needed.
+    this.listGaugeEvictionPolicy.evict(this.elements, this.maxNumberOfItems, this.maxStaleness);
+    return Collections.unmodifiableList(this.elements.stream().map(x -> x.getValue()).collect(Collectors.toList()));
+  }
+
+  /**
+   * Add a value to the list.
+   * (Timestamp assigned to this value is the current timestamp.)
+   * @param value The Gauge value to be added
+   */
+  public void add(T value) {
+    this.elements.add(new TimestampedValue<T>(value, Instant.now().toEpochMilli()));
+
+    // notify the policy object for performing any eviction that may be needed.
+    this.listGaugeEvictionPolicy.evict(this.elements, this.maxNumberOfItems, this.maxStaleness);
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public void visit(MetricsVisitor visitor) {
+    visitor.listGauge(this);
+  }
+}

samza-api/src/main/java/org/apache/samza/metrics/MetricsRegistry.java

@@ -64,6 +64,15 @@ public interface MetricsRegistry {
    */
   <T> Gauge<T> newGauge(String group, Gauge<T> value);
 
+  /**
+   * Register a {@link org.apache.samza.metrics.ListGauge}
+   * @param group Group for this ListGauge
+   * @param listGauge the ListGauge to register
+   * @param <T> Type of the ListGauge
+   * @return ListGauge registered
+   */
+  <T> ListGauge<T> newListGauge(String group, ListGauge<T> listGauge);
+
   /**
    * Create and Register a new {@link org.apache.samza.metrics.Timer}
    * @param group Group for this Timer

samza-api/src/main/java/org/apache/samza/metrics/MetricsVisitor.java

@@ -31,8 +31,13 @@ public abstract class MetricsVisitor {
 
   public abstract void timer(Timer timer);
 
+  public abstract <T> void listGauge(ListGauge<T> listGauge);
+
   public void visit(Metric metric) {
-    if (metric instanceof Counter) {
+    // Cast for metrics of type ListGauge
+    if (metric instanceof ListGauge<?>) {
+      listGauge((ListGauge<?>) metric);
+    } else if (metric instanceof Counter) {
       counter((Counter) metric);
     } else if (metric instanceof Gauge<?>) {
       gauge((Gauge<?>) metric);

samza-api/src/main/java/org/apache/samza/metrics/ReadableMetricsRegistryListener.java

@@ -24,5 +24,7 @@ public interface ReadableMetricsRegistryListener {
 
   void onGauge(String group, Gauge<?> gauge);
 
+  void onListGauge(String group, ListGauge<?> listGauge);
+
   void onTimer(String group, Timer timer);
 }

samza-api/src/main/java/org/apache/samza/util/NoOpMetricsRegistry.java

@@ -21,9 +21,11 @@
 
 import org.apache.samza.metrics.Counter;
 import org.apache.samza.metrics.Gauge;
+import org.apache.samza.metrics.ListGauge;
 import org.apache.samza.metrics.MetricsRegistry;
 import org.apache.samza.metrics.Timer;
 
+
 /**
  * {@link org.apache.samza.metrics.MetricsRegistry} implementation for when no actual metrics need to be
  * recorded but a registry is still required.
@@ -49,6 +51,11 @@ public <T> Gauge<T> newGauge(String group, Gauge<T> gauge) {
     return gauge;
   }
 
+  @Override
+  public <T> ListGauge<T> newListGauge(String group, ListGauge<T> listGauge) {
+    return listGauge;
+  }
+
   @Override
   public Timer newTimer(String group, String name) {
     return new Timer(name);

samza-core/src/main/java/org/apache/samza/operators/impl/store/TimestampedValue.java → samza-api/src/main/java/org/apache/samza/util/TimestampedValue.java

@@ -17,7 +17,7 @@
  * under the License.
  *
  */
-package org.apache.samza.operators.impl.store;
+package org.apache.samza.util;
 
 /**
  * An immutable pair of a value, and its corresponding timestamp.

samza-api/src/test/java/org/apache/samza/metrics/TestListGauge.java

@@ -0,0 +1,99 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.samza.metrics;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.Iterator;
+import org.junit.Assert;
+import org.junit.Test;
+
+
+/**
+ * Class to encapsulate test-cases for {@link org.apache.samza.metrics.ListGauge}
+ */
+public class TestListGauge {
+
+  private final static Duration THREAD_TEST_TIMEOUT = Duration.ofSeconds(10);
+
+  private <T> ListGauge<T> getListGaugeForTest() {
+    return new ListGauge<T>("sampleListGauge", 10, Duration.ofSeconds(60));
+  }
+
+  @Test
+  public void basicTest() {
+    ListGauge<String> listGauge = getListGaugeForTest();
+    listGauge.add("sampleValue");
+    Assert.assertEquals("Names should be the same", listGauge.getName(), "sampleListGauge");
+    Assert.assertEquals("List sizes should match", listGauge.getValues().size(), 1);
+    Assert.assertEquals("ListGauge should contain sampleGauge", listGauge.getValues().contains("sampleValue"), true);
+  }
+
+  @Test
+  public void testSizeEnforcement() {
+    ListGauge listGauge = getListGaugeForTest();
+    for (int i = 15; i > 0; i--) {
+      listGauge.add("v" + i);
+    }
+    Assert.assertEquals("List sizes should be as configured at creation time", listGauge.getValues().size(), 10);
+
+    int valueIndex = 10;
+    Collection<String> currentList = listGauge.getValues();
+    Iterator iterator = currentList.iterator();
+    while (iterator.hasNext()) {
+      String gaugeValue = (String) iterator.next();
+      Assert.assertTrue(gaugeValue.equals("v" + valueIndex));
+      valueIndex--;
+    }
+  }
+
+  @Test
+  public void testThreadSafety() throws InterruptedException {
+    ListGauge<Integer> listGauge = getListGaugeForTest();
+
+    Thread thread1 = new Thread(new Runnable() {
+      @Override
+      public void run() {
+        for (int i = 1; i <= 100; i++) {
+          listGauge.add(i);
+        }
+      }
+    });
+
+    Thread thread2 = new Thread(new Runnable() {
+      @Override
+      public void run() {
+        for (int i = 1; i <= 100; i++) {
+          listGauge.add(i);
+        }
+      }
+    });
+
+    thread1.start();
+    thread2.start();
+
+    thread1.join(THREAD_TEST_TIMEOUT.toMillis());
+    thread2.join(THREAD_TEST_TIMEOUT.toMillis());
+
+    Assert.assertTrue("ListGauge should have the last 10 values", listGauge.getValues().size() == 10);
+    for (Integer gaugeValue : listGauge.getValues()) {
+      Assert.assertTrue("Values should have the last 10 range", gaugeValue <= 100 && gaugeValue > 90);
+    }
+  }
+}

samza-api/src/test/java/org/apache/samza/metrics/TestTimer.java

@@ -62,7 +62,7 @@ public void testTimerWithDifferentWindowSize() {
     assertTrue(snapshot.getValues().containsAll(Arrays.asList(1L, 2L, 3L)));
     assertEquals(3, snapshot.getValues().size());
 
-    // The time is 500 for update(4L) because getSnapshot calls clock once + 3
+    // The time is 500 for update(4L) because getValues calls clock once + 3
     // updates that call clock 3 times
     timer.update(4L);
     Snapshot snapshot2 = timer.getSnapshot();