Merge pull request #194 from brharrington/composite

allow adding and removing from the composite

spectator-api/src/main/java/com/netflix/spectator/api/CompositeCounter.java

@@ -15,34 +15,31 @@
  */
 package com.netflix.spectator.api;
 
+import java.util.Collection;
+import java.util.Iterator;
+
 /** Counter implementation for the composite registry. */
 final class CompositeCounter extends CompositeMeter implements Counter {
 
-  private final Counter[] counters;
-
   /** Create a new instance. */
-  CompositeCounter(Id id, Counter[] counters) {
-    super(id);
-    this.counters = counters;
-  }
-
-  @Override protected Meter[] meters() {
-    return counters;
+  CompositeCounter(Id id, Collection<Registry> registries) {
+    super(id, registries);
   }
 
   @Override public void increment() {
-    for (Counter c : counters) {
-      c.increment();
+    for (Registry r : registries) {
+      r.counter(id).increment();
     }
   }
 
   @Override public void increment(long amount) {
-    for (Counter c : counters) {
-      c.increment(amount);
+    for (Registry r : registries) {
+      r.counter(id).increment(amount);
     }
   }
 
   @Override public long count() {
-    return (counters.length == 0) ? 0L : counters[0].count();
+    Iterator<Registry> it = registries.iterator();
+    return it.hasNext() ? it.next().counter(id).count() : 0L;
   }
 }

spectator-api/src/main/java/com/netflix/spectator/api/CompositeDistributionSummary.java

@@ -15,32 +15,30 @@
  */
 package com.netflix.spectator.api;
 
+import java.util.Collection;
+import java.util.Iterator;
+
 /** Distribution summary implementation for the composite registry. */
 final class CompositeDistributionSummary extends CompositeMeter implements DistributionSummary {
 
-  private final DistributionSummary[] summaries;
-
   /** Create a new instance. */
-  CompositeDistributionSummary(Id id, DistributionSummary[] summaries) {
-    super(id);
-    this.summaries = summaries;
-  }
-
-  @Override protected Meter[] meters() {
-    return summaries;
+  CompositeDistributionSummary(Id id, Collection<Registry> registries) {
+    super(id, registries);
   }
 
   @Override public void record(long amount) {
-    for (DistributionSummary t : summaries) {
-      t.record(amount);
+    for (Registry r : registries) {
+      r.distributionSummary(id).record(amount);
     }
   }
 
   @Override public long count() {
-    return (summaries.length == 0) ? 0L : summaries[0].count();
+    Iterator<Registry> it = registries.iterator();
+    return it.hasNext() ? it.next().distributionSummary(id).count() : 0L;
   }
 
   @Override public long totalAmount() {
-    return (summaries.length == 0) ? 0L : summaries[0].totalAmount();
+    Iterator<Registry> it = registries.iterator();
+    return it.hasNext() ? it.next().distributionSummary(id).totalAmount() : 0L;
   }
 }

spectator-api/src/main/java/com/netflix/spectator/api/CompositeMeter.java

@@ -16,46 +16,52 @@
 package com.netflix.spectator.api;
 
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 
 
 /**
  * Base class for composite implementations of core meter types.
  */
-abstract class CompositeMeter implements Meter {
+class CompositeMeter implements Meter {
 
   /** Identifier for the meter. */
   protected final Id id;
 
+  /** Underlying registries that are keeping the data. */
+  protected final Collection<Registry> registries;
+
   /**
    * Create a new instance.
    *
    * @param id
    *     Identifier for the meter.
    */
-  public CompositeMeter(Id id) {
+  public CompositeMeter(Id id, Collection<Registry> registries) {
     this.id = id;
+    this.registries = registries;
   }
 
-  /** Subclasses should define this method to specify the list of meters for the composite. */
-  protected abstract Meter[] meters();
-
   @Override public Id id() {
     return this.id;
   }
 
   @Override public boolean hasExpired() {
-    for (Meter m : meters()) {
+    for (Registry r : registries) {
+      Meter m = r.get(id);
       if (!m.hasExpired()) return false;
     }
     return true;
   }
 
   @Override public Iterable<Measurement> measure() {
     final List<Measurement> ms = new ArrayList<>();
-    for (Meter meter : meters()) {
-      for (Measurement measurement : meter.measure()) {
-        ms.add(measurement);
+    for (Registry r : registries) {
+      Meter m = r.get(id);
+      if (m != null) {
+        for (Measurement measurement : m.measure()) {
+          ms.add(measurement);
+        }
       }
     }
     return ms;

spectator-api/src/main/java/com/netflix/spectator/api/CompositeRegistry.java

@@ -19,20 +19,31 @@
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.Set;
+import java.util.concurrent.CopyOnWriteArraySet;
 
 /**
  * Maps calls to zero or more sub-registries. If zero then it will act similar to the noop
  * registry. Otherwise activity will be sent to all registries that are part of the composite.
  */
 final class CompositeRegistry implements Registry {
 
+  /**
+   * Id used for a meter storing all gauges registered with the composite. Since there is no
+   * activity on a gauge, all gauges will get registered with a sub-registry when it is added
+   * by including the meter with this id.
+   */
+  static final Id GAUGES_ID = new DefaultId("spectator.composite.gauges");
+
   private final Clock clock;
-  private final Registry[] registries;
+  private final CopyOnWriteArraySet<Registry> registries;
+
+  private final RegistryMeter gauges;
 
   /** Creates a new instance. */
-  CompositeRegistry(Clock clock, Registry[] registries) {
+  CompositeRegistry(Clock clock) {
     this.clock = clock;
-    this.registries = registries;
+    this.registries = new CopyOnWriteArraySet<>();
+    this.gauges = new RegistryMeter(GAUGES_ID);
   }
 
   /**
@@ -49,6 +60,17 @@ <T extends Registry> T find(Class<T> c) {
     return null;
   }
 
+  /** Add a registry to the composite. */
+  public void add(Registry registry) {
+    registries.add(registry);
+    registry.register(gauges);
+  }
+
+  /** Remove a registry from the composite. */
+  public void remove(Registry registry) {
+    registries.remove(registry);
+  }
+
   @Override public Clock clock() {
     return clock;
   }
@@ -62,69 +84,27 @@ <T extends Registry> T find(Class<T> c) {
   }
 
   @Override public void register(Meter meter) {
-    for (Registry r : registries) {
-      r.register(meter);
-    }
+    gauges.register(meter);
   }
 
   @Override public Counter counter(Id id) {
-    final int n = registries.length;
-    if (n == 0) {
-      return NoopCounter.INSTANCE;
-    } else {
-      final Counter[] meters = new Counter[n];
-      for (int i = 0; i < n; ++i) {
-        meters[i] = registries[i].counter(registries[i].createId(id.name(), id.tags()));
-      }
-      return new CompositeCounter(id, meters);
-    }
+    return new CompositeCounter(id, registries);
   }
 
   @Override public DistributionSummary distributionSummary(Id id) {
-    final int n = registries.length;
-    if (n == 0) {
-      return NoopDistributionSummary.INSTANCE;
-    } else {
-      final DistributionSummary[] meters = new DistributionSummary[n];
-      for (int i = 0; i < n; ++i) {
-        meters[i] = registries[i].distributionSummary(registries[i].createId(id.name(), id.tags()));
-      }
-      return new CompositeDistributionSummary(id, meters);
-    }
+    return new CompositeDistributionSummary(id, registries);
   }
 
   @Override public Timer timer(Id id) {
-    final int n = registries.length;
-    if (n == 0) {
-      return NoopTimer.INSTANCE;
-    } else {
-      final Timer[] meters = new Timer[n];
-      for (int i = 0; i < n; ++i) {
-        meters[i] = registries[i].timer(registries[i].createId(id.name(), id.tags()));
-      }
-      return new CompositeTimer(id, clock, meters);
-    }
+    return new CompositeTimer(id, clock, registries);
   }
 
   @Override public Meter get(Id id) {
-    final int n = registries.length;
-    if (n == 0) {
-      return null;
-    } else {
-      final Meter[] meters = new Meter[n];
-      for (int i = 0; i < n; ++i) {
-        meters[i] = registries[i].get(registries[i].createId(id.name(), id.tags()));
-      }
-      return new CompositeMeter(id) {
-        @Override protected Meter[] meters() {
-          return meters;
-        }
-      };
-    }
+    return new CompositeMeter(id, registries);
   }
 
   @Override public Iterator<Meter> iterator() {
-    if (registries.length == 0) {
+    if (registries.isEmpty()) {
       return Collections.<Meter>emptyList().iterator();
     } else {
       final Set<Id> ids = new HashSet<>();

spectator-api/src/main/java/com/netflix/spectator/api/CompositeTimer.java

@@ -15,29 +15,25 @@
  */
 package com.netflix.spectator.api;
 
+import java.util.Collection;
+import java.util.Iterator;
 import java.util.concurrent.Callable;
 import java.util.concurrent.TimeUnit;
 
 /** Timer implementation for the composite registry. */
 final class CompositeTimer extends CompositeMeter implements Timer {
 
   private final Clock clock;
-  private final Timer[] timers;
 
   /** Create a new instance. */
-  CompositeTimer(Id id, Clock clock, Timer[] timers) {
-    super(id);
+  CompositeTimer(Id id, Clock clock, Collection<Registry> registries) {
+    super(id, registries);
     this.clock = clock;
-    this.timers = timers;
-  }
-
-  @Override protected Meter[] meters() {
-    return timers;
   }
 
   @Override public void record(long amount, TimeUnit unit) {
-    for (Timer t : timers) {
-      t.record(amount, unit);
+    for (Registry r : registries) {
+      r.timer(id).record(amount, unit);
     }
   }
 
@@ -62,10 +58,12 @@ final class CompositeTimer extends CompositeMeter implements Timer {
   }
 
   @Override public long count() {
-    return (timers.length == 0) ? 0L : timers[0].count();
+    Iterator<Registry> it = registries.iterator();
+    return it.hasNext() ? it.next().timer(id).count() : 0L;
   }
 
   @Override public long totalTime() {
-    return (timers.length == 0) ? 0L : timers[0].totalTime();
+    Iterator<Registry> it = registries.iterator();
+    return it.hasNext() ? it.next().timer(id).totalTime() : 0L;
   }
 }

spectator-api/src/main/java/com/netflix/spectator/api/RegistryMeter.java

@@ -0,0 +1,65 @@
+/**
+ * Copyright 2015 Netflix, Inc.
+ *
+ * Licensed 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 com.netflix.spectator.api;
+
+import java.util.ArrayList;
+import java.util.List;
+
+
+/**
+ * Wraps a registry as a meter that can be added to another registry. This is currently used to
+ * manage the gauges for the composite registry. Since there is no activity gauges should get
+ * polled and must get tracked as a group and added to each sub-registry.
+ */
+class RegistryMeter implements Meter {
+
+  private final Id id;
+  private final Registry registry;
+
+  /** Create a new instance. */
+  RegistryMeter(Id id) {
+    this(id, new DefaultRegistry());
+  }
+
+  /** Create a new instance. */
+  RegistryMeter(Id id, Registry registry) {
+    this.id = id;
+    this.registry = registry;
+  }
+
+  /** Register a meter with the wrapped registry. */
+  void register(Meter meter) {
+    registry.register(meter);
+  }
+
+  @Override public Id id() {
+    return id;
+  }
+
+  @Override public Iterable<Measurement> measure() {
+    List<Measurement> ms = new ArrayList<>();
+    for (Meter m : registry) {
+      for (Measurement measurement : m.measure()) {
+        ms.add(measurement);
+      }
+    }
+    return ms;
+  }
+
+  @Override public boolean hasExpired() {
+    return false;
+  }
+}