SAMZA-1733 New metric type Listgauge

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

@@ -0,0 +1,105 @@
+/*
+ * 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.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedList;
+import java.util.List;
+
+
+/**
+ * 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 List<T> metricList;
+  private ListGaugeEvictionPolicy<T> listGaugeEvictionPolicy;
+
+  private final static int DEFAULT_POLICY_NUM_RETAIN = 60;
+
+  /**
+   * Create a new {@link ListGauge} with no auto eviction, callers can add/remove items as desired.
+   * @param name Name to be assigned
+   */
+  public ListGauge(String name) {
+    this.name = name;
+    this.metricList = new ArrayList<T>(DEFAULT_POLICY_NUM_RETAIN);
+    this.listGaugeEvictionPolicy = new RetainLastNPolicy<T>(this, DEFAULT_POLICY_NUM_RETAIN);
+  }
+
+  /**
+   * 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.
+   * @return the collection of gauge values
+   */
+  public synchronized Collection<T> getValue() {
+    return Collections.unmodifiableList(this.metricList);
+  }
+
+  /**
+   * Package-private method to change the eviction policy
+   * @param listGaugeEvictionPolicy
+   */
+  synchronized void setEvictionPolicy(ListGaugeEvictionPolicy<T> listGaugeEvictionPolicy) {
+    this.listGaugeEvictionPolicy = listGaugeEvictionPolicy;
+  }
+
+  /**
+   * Add a gauge to the list
+   * @param value The Gauge value to be added
+   */
+  public synchronized void add(T value) {
+    this.metricList.add(value);
+
+    // notify the policy object (if one is present), for performing any eviction that may be needed.
+    // note: monitor is being held
+    if (this.listGaugeEvictionPolicy != null) {
+      this.listGaugeEvictionPolicy.elementAddedCallback();
+    }
+  }
+
+  public synchronized boolean remove(T value) {
+    return this.metricList.remove(value);
+  }
+
+  /**
+   * {@inheritDoc}
+   */
+  @Override
+  public synchronized void visit(MetricsVisitor visitor) {
+    visitor.listGauge(this);
+  }
+}

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

@@ -0,0 +1,23 @@
+/*
+ * 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;
+
+public interface ListGaugeEvictionPolicy<T> {
+  void elementAddedCallback();
+}

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/metrics/RetainLastNPolicy.java

@@ -0,0 +1,47 @@
+/*
+ * 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.util.Collection;
+import java.util.Iterator;
+
+
+public class RetainLastNPolicy<T> implements ListGaugeEvictionPolicy<T> {
+
+  private final ListGauge<T> listGauge;
+  private final int nItems;
+
+  public RetainLastNPolicy(ListGauge<T> listGauge, int numItems) {
+    this.listGauge = listGauge;
+    this.nItems = numItems;
+  }
+
+  @Override
+  public void elementAddedCallback() {
+    // get a snapshot of the list
+    Collection<T> listGaugeCollection = this.listGauge.getValue();
+    int numToEvict = listGaugeCollection.size() - nItems;
+    Iterator<T> iterator = listGaugeCollection.iterator();
+    while (numToEvict > 0 && iterator.hasNext()) {
+      // Remove in FIFO order to retain the last nItems
+      listGauge.remove(iterator.next());
+      numToEvict--;
+    }
+  }
+}

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-api/src/test/java/org/apache/samza/metrics/TestListGauge.java

@@ -0,0 +1,97 @@
+/*
+ * 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);
+
+  @Test
+  public void basicTest() {
+    ListGauge<String> listGauge = new ListGauge<String>("sampleListGauge");
+    listGauge.add("sampleValue");
+    Assert.assertEquals("Names should be the same", listGauge.getName(), "sampleListGauge");
+    Assert.assertEquals("List sizes should match", listGauge.getValue().size(), 1);
+    Assert.assertEquals("ListGauge should contain sampleGauge", listGauge.getValue().contains("sampleValue"), true);
+  }
+
+  @Test
+  public void testSizeEnforcement() {
+    ListGauge listGauge = new ListGauge<String>("listGauge");
+    listGauge.setEvictionPolicy(new RetainLastNPolicy(listGauge, 10));
+    for (int i = 15; i > 0; i--) {
+      listGauge.add("v" + i);
+    }
+    Assert.assertEquals("List sizes should be as configured at creation time", listGauge.getValue().size(), 10);
+
+    int valueIndex = 10;
+    Collection<String> currentList = listGauge.getValue();
+    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 = new ListGauge<Integer>("listGauge");
+    listGauge.setEvictionPolicy(new RetainLastNPolicy(listGauge, 20));
+
+    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 20 values", listGauge.getValue().size() == 20);
+    for (Integer gaugeValue : listGauge.getValue()) {
+      Assert.assertTrue("Values should have the last 20 range", gaugeValue <= 100 && gaugeValue > 80);
+    }
+  }
+}

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 getValue calls clock once + 3
     // updates that call clock 3 times
     timer.update(4L);
     Snapshot snapshot2 = timer.getSnapshot();