You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@commons.apache.org by rm...@apache.org on 2013/10/21 14:54:18 UTC
svn commit: r1534136 [1/2] - in /commons/sandbox/monitoring/trunk: ./
collector/
collector/src/main/java/org/apache/commons/monitoring/collector/rest/
collector/src/main/java/org/apache/commons/monitoring/collector/server/
collector/src/main/java/org/a...
Author: rmannibucau
Date: Mon Oct 21 12:54:16 2013
New Revision: 1534136
URL: http://svn.apache.org/r1534136
Log:
better thread safety + first collector servlet implementation (in memory) + SPI interface (delegating to ServiceLoader by default) to let user override it if needed
Added:
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/
- copied from r1534028, commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/Collector.java
- copied, changed from r1534028, commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/RestCollector.java
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/math/
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/math/Aggregators.java
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/math/M2AwareStatisticalSummary.java
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/AggregatedCollectorCounter.java
- copied, changed from r1534028, commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/store/CollectorCounter.java
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/CollectorCounter.java
- copied, changed from r1534028, commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/store/CollectorCounter.java
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/CollectorCounterStore.java
- copied, changed from r1534028, commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/store/CollectorCounterStore.java
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/InMemoryCollectorCounterStore.java
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/LeafCollectorCounter.java
- copied, changed from r1534028, commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/store/CollectorCounter.java
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/gauge/
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/gauge/CollectorGaugeStore.java
commons/sandbox/monitoring/trunk/collector/src/test/java/org/
commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/
commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/
commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/
commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/
commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/
commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/CollectorServer.java
- copied, changed from r1534028, commons/sandbox/monitoring/trunk/cube/src/test/java/org/apache/commons/monitoring/cube/CubeServer.java
commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/CubeDataStoreCompatibilityTest.java
commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/HttpCollectorTest.java
commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/store/
commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/store/counter/
commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/store/counter/AggregatorTest.java
commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/store/gauge/
commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/store/gauge/CollectorGaugeTest.java
commons/sandbox/monitoring/trunk/collector/src/test/resources/
commons/sandbox/monitoring/trunk/collector/src/test/resources/commons-monitoring.properties
- copied, changed from r1534028, commons/sandbox/monitoring/trunk/cube/src/test/resources/commons-monitoring.properties
commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/spi/
commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/spi/DefaultSPI.java
commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/spi/SPI.java
commons/sandbox/monitoring/trunk/src/site/markdown/collector.md
Removed:
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/RestCollector.java
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/CollectorCounter.java
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/CollectorCounterStore.java
commons/sandbox/monitoring/trunk/collector/src/main/resources/
commons/sandbox/monitoring/trunk/cube/src/main/java/org/apache/commons/monitoring/cube/CubeCounter.java
Modified:
commons/sandbox/monitoring/trunk/collector/pom.xml
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/Event.java
commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/CollectorDataStoreFactory.java
commons/sandbox/monitoring/trunk/core/pom.xml
commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/configuration/Configuration.java
commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/counters/Counter.java
commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/counters/DefaultCounter.java
commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/counters/MetricData.java
commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/gauges/DefaultGaugeManager.java
commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/gauges/Gauge.java
commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/repositories/DefaultRepository.java
commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/store/GaugeDataStore.java
commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/store/InMemoryCounterDataStore.java
commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/store/InMemoryGaugeDataStore.java
commons/sandbox/monitoring/trunk/cube/pom.xml
commons/sandbox/monitoring/trunk/cube/src/main/java/org/apache/commons/monitoring/cube/CubeCounterDataStore.java
commons/sandbox/monitoring/trunk/cube/src/main/java/org/apache/commons/monitoring/cube/CubeDataStoreFactory.java
commons/sandbox/monitoring/trunk/cube/src/main/java/org/apache/commons/monitoring/cube/CubeGaugeDataStore.java
commons/sandbox/monitoring/trunk/cube/src/test/java/org/apache/commons/monitoring/cube/CubeDataStoreTest.java
commons/sandbox/monitoring/trunk/cube/src/test/java/org/apache/commons/monitoring/cube/CubeServer.java
commons/sandbox/monitoring/trunk/graphite/src/main/java/org/apache/commons/monitoring/graphite/GraphiteGaugeDataStore.java
commons/sandbox/monitoring/trunk/graphite/src/test/java/org/apache/commons/monitoring/graphite/GraphiteTest.java
commons/sandbox/monitoring/trunk/pom.xml
commons/sandbox/monitoring/trunk/reporting/src/main/java/org/apache/commons/monitoring/reporting/web/plugin/PluginRepository.java
commons/sandbox/monitoring/trunk/reporting/src/test/java/org/apache/commons/monitoring/reporting/FormatsTest.java
commons/sandbox/monitoring/trunk/src/site/markdown/index.md
Modified: commons/sandbox/monitoring/trunk/collector/pom.xml
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/pom.xml?rev=1534136&r1=1534135&r2=1534136&view=diff
==============================================================================
--- commons/sandbox/monitoring/trunk/collector/pom.xml (original)
+++ commons/sandbox/monitoring/trunk/collector/pom.xml Mon Oct 21 12:54:16 2013
@@ -32,9 +32,7 @@
<dependencies>
<dependency>
<groupId>org.apache.geronimo.specs</groupId>
- <artifactId>geronimo-jaxrs_1.1_spec</artifactId>
- <version>1.0</version>
- <scope>provided</scope>
+ <artifactId>geronimo-servlet_3.0_spec</artifactId>
</dependency>
<dependency>
@@ -49,8 +47,18 @@
<dependency>
+ <groupId>org.apache.commons.monitoring</groupId>
+ <artifactId>commons-monitoring-cube</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
</dependency>
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-codec-http</artifactId>
+ </dependency>
</dependencies>
</project>
Copied: commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/Collector.java (from r1534028, commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/RestCollector.java)
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/Collector.java?p2=commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/Collector.java&p1=commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/RestCollector.java&r1=1534028&r2=1534136&rev=1534136&view=diff
==============================================================================
--- commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/RestCollector.java (original)
+++ commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/Collector.java Mon Oct 21 12:54:16 2013
@@ -14,107 +14,120 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.commons.monitoring.collector.rest;
+package org.apache.commons.monitoring.collector.server;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.monitoring.Role;
-import org.apache.commons.monitoring.collector.rest.store.CollectorCounter;
-import org.apache.commons.monitoring.collector.rest.store.CollectorCounterStore;
+import org.apache.commons.monitoring.collector.server.math.M2AwareStatisticalSummary;
+import org.apache.commons.monitoring.collector.server.store.counter.CollectorCounterStore;
+import org.apache.commons.monitoring.collector.server.store.gauge.CollectorGaugeStore;
import org.apache.commons.monitoring.configuration.Configuration;
import org.apache.commons.monitoring.counters.Counter;
import org.apache.commons.monitoring.counters.Unit;
import org.apache.commons.monitoring.store.CounterDataStore;
import org.apache.commons.monitoring.store.GaugeDataStore;
import org.apache.commons.monitoring.store.InMemoryGaugeDataStore;
+import sun.misc.IOUtils;
-import javax.ws.rs.Consumes;
-import javax.ws.rs.POST;
-import javax.ws.rs.Path;
-import javax.ws.rs.core.MediaType;
-import javax.ws.rs.core.Response;
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServlet;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+import java.net.HttpURLConnection;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.logging.Logger;
+
+// should work with cube clients, see cube module for details
+// Note: for this simple need we don't need JAXRS
+public class Collector extends HttpServlet {
+ private static final Logger LOGGER = Logger.getLogger(Collector.class.getName());
-// close to cube collector API but backed by sirona DataStore
-@Consumes(MediaType.APPLICATION_JSON)
-@Path("/event")
-public class RestCollector {
private static final String OK = "{}";
private static final String GAUGE = "gauge";
private static final String COUNTER = "counter";
- private final CounterDataStore counterDataStore;
- private final GaugeDataStore gaugeDataStore;
+ private final Map<String, Role> roles = new ConcurrentHashMap<String, Role>();
- public RestCollector() {
- gaugeDataStore = Configuration.findOrCreateInstance(GaugeDataStore.class);
- if (!InMemoryGaugeDataStore.class.isInstance(gaugeDataStore)) {
+ private CollectorCounterStore counterDataStore = null;
+ private CollectorGaugeStore gaugeDataStore = null;
+ private ObjectMapper mapper;
+
+ @Override
+ public void init() {
+ final GaugeDataStore gds = Configuration.findOrCreateInstance(GaugeDataStore.class);
+ if (!CollectorGaugeStore.class.isInstance(gds)) {
throw new IllegalStateException("Collector only works with " + InMemoryGaugeDataStore.class.getName());
}
+ this.gaugeDataStore = CollectorGaugeStore.class.cast(gds);
- counterDataStore = Configuration.findOrCreateInstance(CounterDataStore.class);
- if (!CollectorCounterStore.class.isInstance(counterDataStore)) {
+ final CounterDataStore cds = Configuration.findOrCreateInstance(CounterDataStore.class);
+ if (!CollectorCounterStore.class.isInstance(cds)) {
throw new IllegalStateException("Collector only works with " + CollectorCounterStore.class.getName());
}
+ this.counterDataStore = CollectorCounterStore.class.cast(cds);
+
+ this.mapper = new ObjectMapper().configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false).configure(JsonParser.Feature.ALLOW_NON_NUMERIC_NUMBERS, true);
}
- @POST
- @Path("put")
- public Response put(final Event[] events) {
+ @Override
+ protected void doPost(final HttpServletRequest req, final HttpServletResponse resp) throws ServletException, IOException {
+ final Event[] events = mapper.readValue(req.getInputStream(), Event[].class);
if (events != null && events.length > 0) {
try {
- doPut(events);
+ for (final Event event : events) {
+ final String type = event.getType();
+
+ if (COUNTER.equals(type)) {
+ updateCounter(event);
+ } else if (GAUGE.equals(type)) {
+ updateGauge(event);
+ } else {
+ LOGGER.info("Unexpected type '" + type + "', skipping");
+ }
+ }
} catch (final Exception e) {
- return error(e);
+ resp.setStatus(HttpURLConnection.HTTP_BAD_REQUEST);
+ resp.getWriter().write("{\"error\":\"" + e.getMessage().replace('\"', ' ') + "\"}");
+ return;
}
}
- return ok();
- }
- private void doPut(final Event[] events) {
- for (final Event event : events) {
- if (COUNTER.equals(event.getType())) {
- updateCounter(event);
- } else if (GAUGE.equals(event.getType())) {
- updateGauge(event);
- }
- }
+ resp.setStatus(HttpURLConnection.HTTP_OK);
+ resp.getWriter().write(OK);
}
private void updateGauge(final Event event) {
final Map<String,Object> data = event.getData();
final long time = event.getTime().getTime();
-
- final String role = String.class.cast(data.get("role"));
- final String unit = String.class.cast(data.get("unit"));
final double value= Number.class.cast(data.get("value")).doubleValue();
- final Role roleInstance = new Role(role, Unit.get(unit));
- gaugeDataStore.createOrNoopGauge(roleInstance);
- InMemoryGaugeDataStore.class.cast(gaugeDataStore).addToGauge(roleInstance, time, value);
+ gaugeDataStore.addToGauge(role(data), time, value, String.class.cast(data.get("marker")));
}
private void updateCounter(final Event event) {
final Map<String,Object> data = event.getData();
- final long time = event.getTime().getTime();
-
- final long hits = Number.class.cast(data.get("hits")).longValue();
- final long sum = Number.class.cast(data.get("sum")).longValue();
- final int concurrency = Number.class.cast(data.get("concurrency")).intValue();
-
- final String role = String.class.cast(data.get("role"));
- final String unit = String.class.cast(data.get("unit"));
- final String name = String.class.cast(data.get("name"));
-
- final Counter counter = counterDataStore.getOrCreateCounter(new Counter.Key(new Role(role, Unit.get(unit)), name));
- CollectorCounter.class.cast(counter).addEvent(time, hits, sum, concurrency); // we checked the store in the constructor so that's ok
+ counterDataStore.update(
+ new Counter.Key(role(data), String.class.cast(data.get("name"))),
+ String.class.cast(data.get("marker")),
+ new M2AwareStatisticalSummary(data),
+ Number.class.cast(data.get("concurrency")).intValue());
}
- private Response ok() {
- return Response.ok(OK).build();
- }
+ private Role role(final Map<String, Object> data) {
+ final String name = String.class.cast(data.get("role"));
+ final Role existing = roles.get(name);
+ if (existing != null) {
+ return existing;
+ }
- private Response error(final Exception e) {
- return Response.status(400).entity("{\"error\":\"" + e.getMessage() + "\"}").build();
+ final Role created = new Role(name, Unit.get(String.class.cast(data.get("unit"))));
+ roles.put(name, created);
+ return created;
}
}
Modified: commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/Event.java
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/Event.java?rev=1534136&r1=1534028&r2=1534136&view=diff
==============================================================================
--- commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/Event.java (original)
+++ commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/Event.java Mon Oct 21 12:54:16 2013
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.commons.monitoring.collector.rest;
+package org.apache.commons.monitoring.collector.server;
import com.fasterxml.jackson.annotation.JsonFormat;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
@@ -25,7 +25,6 @@ import java.util.Map;
@JsonIgnoreProperties(ignoreUnknown = true)
public class Event {
private String type;
- private String marker; // ?? how to map it in our store ??
private Map<String, Object> data;
@JsonFormat(shape = JsonFormat.Shape.STRING, pattern = "yyyy-MM-dd'T'HH:mm:ss'Z'", timezone = "UTC")
@@ -39,14 +38,6 @@ public class Event {
this.type = type;
}
- public String getMarker() {
- return marker;
- }
-
- public void setMarker(final String marker) {
- this.marker = marker;
- }
-
public Date getTime() {
return time;
}
Added: commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/math/Aggregators.java
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/math/Aggregators.java?rev=1534136&view=auto
==============================================================================
--- commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/math/Aggregators.java (added)
+++ commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/math/Aggregators.java Mon Oct 21 12:54:16 2013
@@ -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.commons.monitoring.collector.server.math;
+
+import org.apache.commons.monitoring.collector.server.store.counter.LeafCollectorCounter;
+
+import java.util.Collection;
+import java.util.Iterator;
+
+public class Aggregators {
+ public static M2AwareStatisticalSummary aggregate(final Collection<LeafCollectorCounter> statistics) {
+ if (statistics == null) {
+ return null;
+ }
+
+ final Iterator<LeafCollectorCounter> iterator = statistics.iterator();
+ if (!iterator.hasNext()) {
+ return null;
+ }
+
+ LeafCollectorCounter current = iterator.next();
+ long n = current.getHits();
+ double min = current.getMin();
+ double sum = current.getSum();
+ double max = current.getMax();
+ double m2 = current.getSecondMoment();
+ double mean = current.getMean();
+ while (iterator.hasNext()) {
+ current = iterator.next();
+ if (current.getMin() < min || Double.isNaN(min)) {
+ min = current.getMin();
+ }
+ if (current.getMax() > max || Double.isNaN(max)) {
+ max = current.getMax();
+ }
+ sum += current.getSum();
+ final double oldN = n;
+ final double curN = current.getHits();
+ n += curN;
+ final double meanDiff = current.getMean() - mean;
+ mean = sum / n;
+ m2 = m2 + current.getSecondMoment() + meanDiff * meanDiff * oldN * curN / n;
+ }
+
+ final double variance;
+ if (n == 0) {
+ variance = Double.NaN;
+ } else if (n == 1) {
+ variance = 0d;
+ } else {
+ variance = m2 / (n - 1);
+ }
+ return new M2AwareStatisticalSummary(mean, variance, n, max, min, sum, m2);
+ }
+
+ private Aggregators() {
+ // no-op
+ }
+}
Added: commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/math/M2AwareStatisticalSummary.java
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/math/M2AwareStatisticalSummary.java?rev=1534136&view=auto
==============================================================================
--- commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/math/M2AwareStatisticalSummary.java (added)
+++ commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/math/M2AwareStatisticalSummary.java Mon Oct 21 12:54:16 2013
@@ -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.commons.monitoring.collector.server.math;
+
+import java.util.Map;
+
+public class M2AwareStatisticalSummary {
+ private final double mean;
+ private final double variance;
+ private final long n;
+ private final double max;
+ private final double min;
+ private final double sum;
+ private final double m2;
+
+ public M2AwareStatisticalSummary(final double mean, final double variance, final long n,
+ final double max, final double min, final double sum,
+ final double m2) {
+ this.mean = mean;
+ this.variance = variance;
+ this.n = n;
+ this.max = max;
+ this.min = min;
+ this.sum = sum;
+ this.m2 = m2;
+ }
+
+ public M2AwareStatisticalSummary(final Map<String, Object> data) {
+ this(toDouble(data.get("mean")), toDouble(data.get("variance")), toLong(data.get("hits")),
+ toDouble(data.get("max")), toDouble(data.get("min")), toDouble(data.get("sum")),
+ toDouble(data.get("m2")));
+ }
+
+ private static double toDouble(final Object mean) {
+ if (Number.class.isInstance(mean)) {
+ return Number.class.cast(mean).doubleValue();
+ }
+ if (String.class.isInstance(mean)) {
+ return Double.parseDouble(String.class.cast(mean));
+ }
+ throw new IllegalArgumentException(mean + " not supported");
+ }
+
+ private static long toLong(final Object mean) {
+ if (Number.class.isInstance(mean)) {
+ return Number.class.cast(mean).longValue();
+ }
+ if (String.class.isInstance(mean)) {
+ return Long.parseLong(String.class.cast(mean));
+ }
+ throw new IllegalArgumentException(mean + " not supported");
+ }
+
+ public double getSecondMoment() {
+ return m2;
+ }
+
+ public double getMean() {
+ return mean;
+ }
+
+ public double getVariance() {
+ return variance;
+ }
+
+ public long getN() {
+ return n;
+ }
+
+ public double getMax() {
+ return max;
+ }
+
+ public double getMin() {
+ return min;
+ }
+
+ public double getSum() {
+ return sum;
+ }
+
+ public double getM2() {
+ return m2;
+ }
+}
Modified: commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/CollectorDataStoreFactory.java
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/CollectorDataStoreFactory.java?rev=1534136&r1=1534028&r2=1534136&view=diff
==============================================================================
--- commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/CollectorDataStoreFactory.java (original)
+++ commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/CollectorDataStoreFactory.java Mon Oct 21 12:54:16 2013
@@ -14,13 +14,14 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.commons.monitoring.collector.rest.store;
+package org.apache.commons.monitoring.collector.server.store;
+import org.apache.commons.monitoring.collector.server.store.counter.InMemoryCollectorCounterStore;
+import org.apache.commons.monitoring.collector.server.store.gauge.CollectorGaugeStore;
import org.apache.commons.monitoring.store.DelegateDataStoreFactory;
-import org.apache.commons.monitoring.store.InMemoryGaugeDataStore;
public class CollectorDataStoreFactory extends DelegateDataStoreFactory {
public CollectorDataStoreFactory() {
- super(new CollectorCounterStore(), new InMemoryGaugeDataStore());
+ super(new InMemoryCollectorCounterStore(), new CollectorGaugeStore());
}
}
Copied: commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/AggregatedCollectorCounter.java (from r1534028, commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/store/CollectorCounter.java)
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/AggregatedCollectorCounter.java?p2=commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/AggregatedCollectorCounter.java&p1=commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/store/CollectorCounter.java&r1=1534028&r2=1534136&rev=1534136&view=diff
==============================================================================
--- commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/store/CollectorCounter.java (original)
+++ commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/AggregatedCollectorCounter.java Mon Oct 21 12:54:16 2013
@@ -14,34 +14,44 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.commons.monitoring.collector.rest.store;
+package org.apache.commons.monitoring.collector.server.store.counter;
-import org.apache.commons.monitoring.counters.DefaultCounter;
-import org.apache.commons.monitoring.store.CounterDataStore;
+import org.apache.commons.monitoring.collector.server.math.Aggregators;
-public class CollectorCounter extends DefaultCounter {
- public CollectorCounter(final Key key, final CounterDataStore store) {
- super(key, store);
- }
+import java.util.Collection;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.locks.Lock;
- public void addEvent(final long time, final long hits, final long sum, final int concurrency) {
- if (hits == 0) {
- return;
- }
+public class AggregatedCollectorCounter extends CollectorCounter {
+ private final ConcurrentMap<String, LeafCollectorCounter> aggregation = new ConcurrentHashMap<String, LeafCollectorCounter>(50);
- // TODO: find a better solution? to let hits be correct we consider we add N times the average which is
- // mathematically wrong
- // a best solution would be to push all raw data
- // but it has big impact on the measure side that we don't want
- final double avg = sum * 1. / hits;
- lock.lock();
+ public AggregatedCollectorCounter(Key key) {
+ super(key);
+ }
+
+ public void update() {
+ final Lock workLock = lock.writeLock();
+ workLock.lock();
try {
- for (long i = 0; i < hits; i++) {
- addInternal(avg);
- }
- updateConcurrency(concurrency);
+ final Collection<LeafCollectorCounter> counters = aggregation.values();
+ statistics = Aggregators.aggregate(counters);
+ concurrency.set(computeConcurrency(counters));
+ updateConcurrency(concurrency.get());
} finally {
- lock.unlock();
+ workLock.unlock();
+ }
+ }
+
+ public void addIfMissing(final String marker, final LeafCollectorCounter counter) {
+ aggregation.putIfAbsent(marker, counter);
+ }
+
+ private static int computeConcurrency(final Collection<LeafCollectorCounter> counters) {
+ int i = 0;
+ for (final LeafCollectorCounter counter : counters) {
+ i += counter.currentConcurrency().get();
}
+ return i;
}
}
Copied: commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/CollectorCounter.java (from r1534028, commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/store/CollectorCounter.java)
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/CollectorCounter.java?p2=commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/CollectorCounter.java&p1=commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/store/CollectorCounter.java&r1=1534028&r2=1534136&rev=1534136&view=diff
==============================================================================
--- commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/store/CollectorCounter.java (original)
+++ commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/CollectorCounter.java Mon Oct 21 12:54:16 2013
@@ -14,34 +14,169 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.commons.monitoring.collector.rest.store;
+package org.apache.commons.monitoring.collector.server.store.counter;
-import org.apache.commons.monitoring.counters.DefaultCounter;
-import org.apache.commons.monitoring.store.CounterDataStore;
+import org.apache.commons.monitoring.collector.server.math.M2AwareStatisticalSummary;
+import org.apache.commons.monitoring.counters.Counter;
+import org.apache.commons.monitoring.counters.Unit;
-public class CollectorCounter extends DefaultCounter {
- public CollectorCounter(final Key key, final CounterDataStore store) {
- super(key, store);
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+public abstract class CollectorCounter implements Counter {
+ protected final Key key;
+ protected final ReadWriteLock lock = new ReentrantReadWriteLock();
+ protected volatile int maxConcurrency = 0;
+ protected volatile AtomicInteger concurrency = new AtomicInteger(0);
+ protected M2AwareStatisticalSummary statistics;
+
+ public CollectorCounter(final Key key) {
+ this.key = key;
+ reset();
}
- public void addEvent(final long time, final long hits, final long sum, final int concurrency) {
- if (hits == 0) {
- return;
- }
+ @Override
+ public Key getKey() {
+ return key;
+ }
- // TODO: find a better solution? to let hits be correct we consider we add N times the average which is
- // mathematically wrong
- // a best solution would be to push all raw data
- // but it has big impact on the measure side that we don't want
- final double avg = sum * 1. / hits;
- lock.lock();
+ @Override
+ public void reset() {
+ final Lock workLock = lock.writeLock();
+ workLock.lock();
try {
- for (long i = 0; i < hits; i++) {
- addInternal(avg);
+ statistics = new M2AwareStatisticalSummary(Double.NaN, Double.NaN, 0, Double.NaN, Double.NaN, Double.NaN, Double.NaN);
+ } finally {
+ workLock.unlock();
+ }
+ }
+
+ @Override
+ public void add(final double delta) {
+ // no-op: this counter is updated through update method
+ }
+
+ @Override
+ public void add(final double delta, Unit unit) {
+ add(key.getRole().getUnit().convert(delta, unit));
+ }
+
+ @Override
+ public AtomicInteger currentConcurrency() {
+ return concurrency;
+ }
+
+ @Override
+ public void updateConcurrency(final int concurrency) {
+ if (concurrency > maxConcurrency) {
+ final Lock workLock = lock.writeLock();
+ workLock.lock();
+ try {
+ maxConcurrency = concurrency;
+ } finally {
+ workLock.unlock();
}
- updateConcurrency(concurrency);
+ }
+ }
+
+ @Override
+ public int getMaxConcurrency() {
+ final Lock workLock = lock.readLock();
+ workLock.lock();
+ try {
+ return maxConcurrency;
+ } finally {
+ workLock.unlock();
+ }
+ }
+
+ @Override
+ public double getMax() {
+ final Lock workLock = lock.readLock();
+ workLock.lock();
+ try {
+ return statistics.getMax();
+ } finally {
+ workLock.unlock();
+ }
+ }
+
+ @Override
+ public double getMin() {
+ final Lock workLock = lock.readLock();
+ workLock.lock();
+ try {
+ return statistics.getMin();
+ } finally {
+ workLock.unlock();
+ }
+ }
+
+ @Override
+ public long getHits() {
+ final Lock workLock = lock.readLock();
+ workLock.lock();
+ try {
+ return statistics.getN();
+ } finally {
+ workLock.unlock();
+ }
+ }
+
+ @Override
+ public double getSum() {
+ final Lock workLock = lock.readLock();
+ workLock.lock();
+ try {
+ return statistics.getSum();
+ } finally {
+ workLock.unlock();
+ }
+ }
+
+ @Override
+ public double getStandardDeviation() {
+ final Lock workLock = lock.readLock();
+ workLock.lock();
+ try {
+ return Math.sqrt(statistics.getVariance());
+ } finally {
+ workLock.unlock();
+ }
+ }
+
+ @Override
+ public double getVariance() {
+ final Lock workLock = lock.readLock();
+ workLock.lock();
+ try {
+ return statistics.getVariance();
+ } finally {
+ workLock.unlock();
+ }
+ }
+
+ @Override
+ public double getMean() {
+ final Lock workLock = lock.readLock();
+ workLock.lock();
+ try {
+ return statistics.getMean();
+ } finally {
+ workLock.unlock();
+ }
+ }
+
+ @Override
+ public double getSecondMoment() {
+ final Lock workLock = lock.readLock();
+ workLock.lock();
+ try {
+ return statistics.getSecondMoment();
} finally {
- lock.unlock();
+ workLock.unlock();
}
}
}
Copied: commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/CollectorCounterStore.java (from r1534028, commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/store/CollectorCounterStore.java)
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/CollectorCounterStore.java?p2=commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/CollectorCounterStore.java&p1=commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/store/CollectorCounterStore.java&r1=1534028&r2=1534136&rev=1534136&view=diff
==============================================================================
--- commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/store/CollectorCounterStore.java (original)
+++ commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/CollectorCounterStore.java Mon Oct 21 12:54:16 2013
@@ -14,14 +14,17 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.commons.monitoring.collector.rest.store;
+package org.apache.commons.monitoring.collector.server.store.counter;
+import org.apache.commons.monitoring.collector.server.math.M2AwareStatisticalSummary;
import org.apache.commons.monitoring.counters.Counter;
-import org.apache.commons.monitoring.store.InMemoryCounterDataStore;
+import org.apache.commons.monitoring.store.CounterDataStore;
-public class CollectorCounterStore extends InMemoryCounterDataStore {
- @Override
- protected Counter newCounter(final Counter.Key key) {
- return new CollectorCounter(key, this);
- }
+import java.util.Collection;
+
+public interface CollectorCounterStore extends CounterDataStore {
+ void update(Counter.Key key, String marker, M2AwareStatisticalSummary stats, int concurrency);
+ Collection<String> markers();
+ Collection<? extends Counter> getCounters(String marker);
+ LeafCollectorCounter getOrCreateCounter(Counter.Key key, final String marker);
}
Added: commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/InMemoryCollectorCounterStore.java
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/InMemoryCollectorCounterStore.java?rev=1534136&view=auto
==============================================================================
--- commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/InMemoryCollectorCounterStore.java (added)
+++ commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/InMemoryCollectorCounterStore.java Mon Oct 21 12:54:16 2013
@@ -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.commons.monitoring.collector.server.store.counter;
+
+import org.apache.commons.monitoring.collector.server.math.M2AwareStatisticalSummary;
+import org.apache.commons.monitoring.counters.Counter;
+import org.apache.commons.monitoring.store.InMemoryCounterDataStore;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+public class InMemoryCollectorCounterStore extends InMemoryCounterDataStore implements CollectorCounterStore {
+ private final ConcurrentMap<String, ConcurrentMap<Counter.Key, LeafCollectorCounter>> countersByMarker = new ConcurrentHashMap<String, ConcurrentMap<Counter.Key, LeafCollectorCounter>>();
+
+ @Override
+ public void update(final Counter.Key key, final String marker, final M2AwareStatisticalSummary stats, final int concurrency) {
+ getOrCreateCounter(key, marker).update(stats, concurrency);
+ getOrCreateCounter(key).update();
+ }
+
+ @Override
+ public Collection<String> markers() {
+ return countersByMarker.keySet();
+ }
+
+ @Override
+ public Collection<? extends Counter> getCounters(final String marker) {
+ return countersByMarker.get(marker).values();
+ }
+
+ @Override
+ public LeafCollectorCounter getOrCreateCounter(final Counter.Key key, final String marker) {
+ ConcurrentMap<Counter.Key, LeafCollectorCounter> subCounters = countersByMarker.get(marker);
+ if (subCounters == null) {
+ final ConcurrentMap<Counter.Key, LeafCollectorCounter> map = new ConcurrentHashMap<Counter.Key, LeafCollectorCounter>(50);
+ final ConcurrentMap<Counter.Key, LeafCollectorCounter> existing = countersByMarker.putIfAbsent(marker, map);
+ if (existing != null) {
+ subCounters = existing;
+ } else {
+ subCounters = map;
+ }
+ }
+
+ LeafCollectorCounter counter = subCounters.get(key);
+ if (counter == null) {
+ counter = new LeafCollectorCounter(key);
+ final LeafCollectorCounter previous = subCounters.putIfAbsent(key, counter);
+ if (previous != null) {
+ counter = previous;
+ }
+ final AggregatedCollectorCounter aggregate = AggregatedCollectorCounter.class.cast(super.getOrCreateCounter(key));
+ aggregate.addIfMissing(marker, counter);
+ }
+
+ return counter;
+ }
+
+ @Override
+ protected Counter newCounter(final Counter.Key key) {
+ return new AggregatedCollectorCounter(key);
+ }
+
+ @Override
+ public AggregatedCollectorCounter getOrCreateCounter(final Counter.Key key) {
+ return AggregatedCollectorCounter.class.cast(super.getOrCreateCounter(key));
+ }
+
+ @Override
+ public void clearCounters() {
+ for (final Map.Entry<String, ConcurrentMap<Counter.Key, LeafCollectorCounter>> maps : countersByMarker.entrySet()) {
+ maps.getValue().clear();
+ }
+ countersByMarker.clear();
+ super.clearCounters();
+ }
+
+ @Override
+ public void addToCounter(final Counter defaultCounter, final double delta) {
+ throw new UnsupportedOperationException("shouldn't be used");
+ }
+}
Copied: commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/LeafCollectorCounter.java (from r1534028, commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/store/CollectorCounter.java)
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/LeafCollectorCounter.java?p2=commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/LeafCollectorCounter.java&p1=commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/store/CollectorCounter.java&r1=1534028&r2=1534136&rev=1534136&view=diff
==============================================================================
--- commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/rest/store/CollectorCounter.java (original)
+++ commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/counter/LeafCollectorCounter.java Mon Oct 21 12:54:16 2013
@@ -14,34 +14,27 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.commons.monitoring.collector.rest.store;
+package org.apache.commons.monitoring.collector.server.store.counter;
-import org.apache.commons.monitoring.counters.DefaultCounter;
-import org.apache.commons.monitoring.store.CounterDataStore;
+import org.apache.commons.monitoring.collector.server.math.M2AwareStatisticalSummary;
+import org.apache.commons.monitoring.counters.Counter;
-public class CollectorCounter extends DefaultCounter {
- public CollectorCounter(final Key key, final CounterDataStore store) {
- super(key, store);
- }
+import java.util.concurrent.locks.Lock;
- public void addEvent(final long time, final long hits, final long sum, final int concurrency) {
- if (hits == 0) {
- return;
- }
+public class LeafCollectorCounter extends CollectorCounter {
+ public LeafCollectorCounter(final Counter.Key key) {
+ super(key);
+ }
- // TODO: find a better solution? to let hits be correct we consider we add N times the average which is
- // mathematically wrong
- // a best solution would be to push all raw data
- // but it has big impact on the measure side that we don't want
- final double avg = sum * 1. / hits;
- lock.lock();
+ public void update(final M2AwareStatisticalSummary newStats, final int newConcurrency) {
+ final Lock workLock = lock.writeLock();
+ workLock.lock();
try {
- for (long i = 0; i < hits; i++) {
- addInternal(avg);
- }
- updateConcurrency(concurrency);
+ concurrency.set(newConcurrency);
+ updateConcurrency(newConcurrency);
+ statistics = newStats;
} finally {
- lock.unlock();
+ workLock.unlock();
}
}
}
Added: commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/gauge/CollectorGaugeStore.java
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/gauge/CollectorGaugeStore.java?rev=1534136&view=auto
==============================================================================
--- commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/gauge/CollectorGaugeStore.java (added)
+++ commons/sandbox/monitoring/trunk/collector/src/main/java/org/apache/commons/monitoring/collector/server/store/gauge/CollectorGaugeStore.java Mon Oct 21 12:54:16 2013
@@ -0,0 +1,121 @@
+/*
+ * 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.commons.monitoring.collector.server.store.gauge;
+
+import org.apache.commons.monitoring.MonitoringException;
+import org.apache.commons.monitoring.Role;
+import org.apache.commons.monitoring.configuration.Configuration;
+import org.apache.commons.monitoring.store.GaugeDataStore;
+import org.apache.commons.monitoring.store.GaugeValuesRequest;
+import org.apache.commons.monitoring.store.InMemoryGaugeDataStore;
+
+import java.lang.reflect.Constructor;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+public class CollectorGaugeStore implements GaugeDataStore {
+ private final ConcurrentMap<String, GaugeDataStore> dataStores = new ConcurrentHashMap<String, GaugeDataStore>();
+
+ private final Class<? extends GaugeDataStore> delegateClass;
+
+ public CollectorGaugeStore() {
+ try {
+ delegateClass = Class.class.cast(
+ CollectorGaugeStore.class.getClassLoader().loadClass( // use this classloader and not TCCL to avoid issues
+ Configuration.getProperty(Configuration.COMMONS_MONITORING_PREFIX + "collector.gauge.store-class", InMemoryGaugeDataStore.class.getName())));
+ } catch (final ClassNotFoundException e) {
+ throw new MonitoringException(e);
+ }
+ }
+
+ protected GaugeDataStore newStore(final String marker) {
+ try {
+ try {
+ final Constructor<? extends GaugeDataStore> cons = delegateClass.getConstructor(String.class);
+ return cons.newInstance(marker);
+ } catch (final Exception e) {
+ // no-op: use default constructor
+ }
+ return delegateClass.newInstance();
+ } catch (final Exception e) {
+ throw new MonitoringException(e);
+ }
+ }
+
+ public Map<Long, Double> getGaugeValues(final GaugeValuesRequest gaugeValuesRequest, final String marker) {
+ final GaugeDataStore gaugeDataStore = dataStores.get(marker);
+ if (gaugeDataStore == null) {
+ return Collections.emptyMap();
+ }
+ return gaugeDataStore.getGaugeValues(gaugeValuesRequest);
+ }
+
+ public void createOrNoopGauge(final Role role, final String marker) {
+ GaugeDataStore gaugeDataStore = dataStores.get(marker);
+ if (gaugeDataStore == null) {
+ gaugeDataStore = newStore(marker);
+ final GaugeDataStore existing = dataStores.putIfAbsent(marker, gaugeDataStore);
+ if (existing != null) {
+ gaugeDataStore = existing;
+ }
+ }
+ gaugeDataStore.createOrNoopGauge(role);
+ }
+
+ public void addToGauge(final Role role, final long time, final double value, final String marker) {
+ createOrNoopGauge(role, marker); // this implementation doesn't mandates createOrNoopGauge call
+ dataStores.get(marker).addToGauge(role, time, value);
+ }
+
+ public Collection<String> markers() {
+ return dataStores.keySet();
+ }
+
+ @Override
+ public Map<Long, Double> getGaugeValues(final GaugeValuesRequest gaugeValuesRequest) {
+ final Map<Long, Double> values = new HashMap<Long, Double>();
+ for (final Map.Entry<String, GaugeDataStore> marker : dataStores.entrySet()) {
+ final Map<Long, Double> gaugeValues = marker.getValue().getGaugeValues(gaugeValuesRequest);
+ for (final Map.Entry<Long, Double> entry : gaugeValues.entrySet()) {
+ final Long key = entry.getKey();
+ final Double value = values.get(key);
+ final Double thisValue = entry.getValue();
+ if (value == null) {
+ values.put(key, thisValue);
+ } else {
+ values.put(key, value + thisValue);
+ }
+ }
+ }
+
+ return values;
+ }
+
+ @Override
+ public void createOrNoopGauge(final Role role) {
+ throw new UnsupportedOperationException("Need a marker");
+ }
+
+ @Override
+ public void addToGauge(final Role role, final long time, final double value) {
+ throw new UnsupportedOperationException("Need a marker");
+ }
+}
Copied: commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/CollectorServer.java (from r1534028, commons/sandbox/monitoring/trunk/cube/src/test/java/org/apache/commons/monitoring/cube/CubeServer.java)
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/CollectorServer.java?p2=commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/CollectorServer.java&p1=commons/sandbox/monitoring/trunk/cube/src/test/java/org/apache/commons/monitoring/cube/CubeServer.java&r1=1534028&r2=1534136&rev=1534136&view=diff
==============================================================================
--- commons/sandbox/monitoring/trunk/cube/src/test/java/org/apache/commons/monitoring/cube/CubeServer.java (original)
+++ commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/CollectorServer.java Mon Oct 21 12:54:16 2013
@@ -14,7 +14,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.commons.monitoring.cube;
+package org.apache.commons.monitoring.collector.server;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
@@ -32,30 +32,36 @@ import io.netty.handler.codec.http.FullH
import io.netty.handler.codec.http.HttpMethod;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpRequestDecoder;
-import io.netty.handler.codec.http.HttpResponse;
import io.netty.handler.codec.http.HttpResponseEncoder;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.stream.ChunkedWriteHandler;
+import javax.servlet.ServletInputStream;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
+import java.io.InputStream;
+import java.io.PrintWriter;
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Method;
+import java.lang.reflect.Proxy;
import java.net.ServerSocket;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.LinkedList;
+import java.nio.charset.Charset;
import java.util.logging.Level;
import java.util.logging.Logger;
-public class CubeServer {
- private static final Logger LOGGER = Logger.getLogger(CubeServer.class.getName());
+public class CollectorServer {
+ private static final Logger LOGGER = Logger.getLogger(CollectorServer.class.getName());
private final String host;
private final int port;
private NioEventLoopGroup workerGroup;
- private final Collection<String> messages = new LinkedList<String>();
- public CubeServer(final String host, final int port) {
+ public CollectorServer(final String host, final int port) {
this.host = host;
if (port <= 0) { // generate a port
this.port = findNextAvailablePort();
@@ -63,13 +69,6 @@ public class CubeServer {
this.port = port;
}
}
-
- public Collection<String> getMessages() {
- synchronized (messages) {
- return new ArrayList<String>(messages);
- }
- }
-
public int getPort() {
return port;
}
@@ -93,7 +92,7 @@ public class CubeServer {
return 0;
}
- public CubeServer start() {
+ public CollectorServer start() {
workerGroup = new NioEventLoopGroup(8);
try {
@@ -104,7 +103,7 @@ public class CubeServer {
.option(ChannelOption.TCP_NODELAY, true)
.group(workerGroup)
.channel(NioServerSocketChannel.class)
- .childHandler(new Initializer(messages))
+ .childHandler(new Initializer())
.bind(host, port).addListener(new ChannelFutureListener() {
@Override
public void operationComplete(final ChannelFuture future) throws Exception {
@@ -130,12 +129,6 @@ public class CubeServer {
}
private static class Initializer extends ChannelInitializer<SocketChannel> {
- private final Collection<String> messages;
-
- private Initializer(final Collection<String> messages) {
- this.messages = messages;
- }
-
@Override
protected void initChannel(final SocketChannel ch) throws Exception {
final ChannelPipeline pipeline = ch.pipeline();
@@ -145,25 +138,67 @@ public class CubeServer {
.addLast("aggregator", new HttpObjectAggregator(Integer.MAX_VALUE))
.addLast("encoder", new HttpResponseEncoder())
.addLast("chunked-writer", new ChunkedWriteHandler())
- .addLast("featured-mock-server", new RequestHandler(messages));
+ .addLast("featured-mock-server", new RequestHandler());
}
}
private static class RequestHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
- private final Collection<String> messages;
+ private final Collector collector;
- private RequestHandler(final Collection<String> messages) {
- this.messages = messages;
+ private RequestHandler() {
+ collector = new Collector();
+ collector.init();
}
@Override
protected void channelRead0(final ChannelHandlerContext ctx, final FullHttpRequest fullHttpRequest) throws Exception {
final ChannelFuture future;
if (HttpMethod.POST.equals(fullHttpRequest.getMethod())) {
- synchronized (messages) {
- messages.add(new String(fullHttpRequest.content().array()));
- }
- final HttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
+ final InputStream is = new ByteArrayInputStream(fullHttpRequest.content().toString(Charset.defaultCharset()).getBytes());
+
+ final DefaultFullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK);
+ final ByteArrayOutputStream baos = new ByteArrayOutputStream();
+ final PrintWriter writer = new PrintWriter(baos);
+
+ collector.doPost(HttpServletRequest.class.cast(Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[]{HttpServletRequest.class}, new InvocationHandler() {
+ @Override
+ public Object invoke(final Object proxy, final Method method, final Object[] args) throws Throwable {
+ if (Object.class.equals(method.getDeclaringClass())) {
+ return method.invoke(this, args);
+ }
+
+ if ("getInputStream".equals(method.getName())) {
+ return new ServletInputStream() {
+ @Override
+ public int read() throws IOException {
+ return is.read();
+ }
+ };
+ }
+
+ throw new UnsupportedOperationException("not implemented");
+ }
+ })),
+ HttpServletResponse.class.cast(Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[] { HttpServletResponse.class}, new InvocationHandler() {
+ @Override
+ public Object invoke(final Object proxy, final Method method, final Object[] args) throws Throwable {
+ if (Object.class.equals(method.getDeclaringClass())) {
+ return method.invoke(this, args);
+ }
+
+ final String name = method.getName();
+ if ("setStatus".equals(name)) {
+ response.setStatus(HttpResponseStatus.valueOf(Integer.class.cast(args[0])));
+ return null;
+ } else if ("getWriter".equals(name)) {
+ return writer;
+ }
+
+ throw new UnsupportedOperationException("not implemented");
+ }
+ })));
+
+ response.content().writeBytes(baos.toByteArray());
future = ctx.writeAndFlush(response);
} else {
LOGGER.warning("Received " + fullHttpRequest.getMethod());
Added: commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/CubeDataStoreCompatibilityTest.java
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/CubeDataStoreCompatibilityTest.java?rev=1534136&view=auto
==============================================================================
--- commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/CubeDataStoreCompatibilityTest.java (added)
+++ commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/CubeDataStoreCompatibilityTest.java Mon Oct 21 12:54:16 2013
@@ -0,0 +1,63 @@
+/*
+ * 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.commons.monitoring.collector.server;
+
+import org.apache.commons.monitoring.Role;
+import org.apache.commons.monitoring.collector.server.store.counter.CollectorCounterStore;
+import org.apache.commons.monitoring.configuration.Configuration;
+import org.apache.commons.monitoring.counters.Counter;
+import org.apache.commons.monitoring.counters.Unit;
+import org.apache.commons.monitoring.cube.CubeCounterDataStore;
+import org.apache.commons.monitoring.repositories.Repository;
+import org.apache.commons.monitoring.store.CounterDataStore;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+public class CubeDataStoreCompatibilityTest {
+ private CollectorServer server;
+
+ @Before
+ public void start() {
+ server = new CollectorServer("localhost", 1234).start();
+ Repository.INSTANCE.clear();
+ }
+
+ @After
+ public void shutdown() {
+ server.stop();
+ Repository.INSTANCE.clear();
+ }
+
+ @Test
+ public void cubeMe() {
+ Repository.INSTANCE.getCounter(new Counter.Key(new Role("cube", Unit.UNARY), "client")).add(50);
+ final SeeMyProtectedStuffStore ccds = new SeeMyProtectedStuffStore();
+ ccds.doPush();
+
+ final CollectorCounterStore store = CollectorCounterStore.class.cast(Configuration.getInstance(CounterDataStore.class));
+ final Counter counter1 = store.getOrCreateCounter(new Counter.Key(new Role("cube", Unit.UNARY), "client"));
+ final Counter counter1Client1 = store.getOrCreateCounter(new Counter.Key(new Role("cube", Unit.UNARY), "client"), "local");
+
+ }
+
+ private static class SeeMyProtectedStuffStore extends CubeCounterDataStore {
+ public void doPush() {
+ pushCountersByBatch(Repository.INSTANCE);
+ }
+ }
+}
Added: commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/HttpCollectorTest.java
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/HttpCollectorTest.java?rev=1534136&view=auto
==============================================================================
--- commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/HttpCollectorTest.java (added)
+++ commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/HttpCollectorTest.java Mon Oct 21 12:54:16 2013
@@ -0,0 +1,165 @@
+/*
+ * 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.commons.monitoring.collector.server;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.monitoring.Role;
+import org.apache.commons.monitoring.collector.server.store.counter.CollectorCounterStore;
+import org.apache.commons.monitoring.configuration.Configuration;
+import org.apache.commons.monitoring.counters.Counter;
+import org.apache.commons.monitoring.counters.Unit;
+import org.apache.commons.monitoring.repositories.Repository;
+import org.apache.commons.monitoring.store.CounterDataStore;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.StringWriter;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+public class HttpCollectorTest {
+ private CollectorServer server;
+ private ObjectMapper mapper;
+
+ @Before
+ public void start() {
+ server = new CollectorServer("localhost", 1234).start();
+ mapper = new ObjectMapper();
+ Repository.INSTANCE.clear();
+ }
+
+ @After
+ public void shutdown() {
+ server.stop();
+ Repository.INSTANCE.clear();
+ }
+
+ @Test
+ public void collect() throws Exception {
+ {
+ final Event[] events1 = new Event[2];
+ {
+ events1[0] = new Event();
+ events1[0].setType("counter");
+ events1[0].setData(buildData("counter1", "role1", Unit.UNARY.getName(), "client1", 6, 8, 7, 10, 12, 4, 64, 55));
+ }
+ {
+ events1[1] = new Event();
+ events1[1].setType("counter");
+ events1[1].setData(buildData("counter2", "role2", Unit.UNARY.getName(), "client1", 8, 10, 3, 156, 75, 44, 4, 525));
+ }
+ doPost(events1);
+ }
+
+ {
+ final Event[] events2 = new Event[2];
+ {
+ events2[0] = new Event();
+ events2[0].setType("counter");
+ events2[0].setData(buildData("counter1", "role1", Unit.UNARY.getName(), "client2", 7, 64, 78, 190, 612, 46, 654, 5));
+ }
+ {
+ events2[1] = new Event();
+ events2[1].setType("counter");
+ events2[1].setData(buildData("counter2", "role2", Unit.UNARY.getName(), "client2", 84, 10978, 3869, 1586, 715, 474, 44, 65));
+ }
+ doPost(events2);
+ }
+
+ final CollectorCounterStore store = CollectorCounterStore.class.cast(Configuration.getInstance(CounterDataStore.class));
+ final Counter counter1 = store.getOrCreateCounter(new Counter.Key(new Role("role1", Unit.UNARY), "counter1"));
+ final Counter counter1Client1 = store.getOrCreateCounter(new Counter.Key(new Role("role1", Unit.UNARY), "counter1"), "client1");
+ final Counter counter1Client2 = store.getOrCreateCounter(new Counter.Key(new Role("role1", Unit.UNARY), "counter1"), "client2");
+ assertCounter(counter1, 200, 4, 612, 3.59, 12.24785, 150.01005, 718);
+ assertCounter(counter1Client1, 10, 4, 12, 8, 2.64575, 7, 64);
+ assertCounter(counter1Client2, 190, 46, 612, 64, 8.83176, 78, 654);
+ }
+
+ private void doPost(final Event[] events) throws Exception {
+ final URL url = new URL("http://localhost:" + server.getPort());
+
+ final HttpURLConnection connection = HttpURLConnection.class.cast(url.openConnection());
+ connection.setRequestMethod("POST");
+ connection.setUseCaches(false);
+ connection.setDoInput(true);
+ connection.setDoOutput(true);
+
+ final StringWriter writer = new StringWriter();
+ mapper.writeValue(writer, events);
+
+ try {
+ final OutputStream output = connection.getOutputStream();
+ try {
+ output.write(writer.toString().getBytes());
+ output.flush();
+
+ final int status = connection.getResponseCode();
+ if (status / 100 != 2) {
+ throw new IOException("Status = " + status);
+ }
+ } finally {
+ if (output != null) {
+ output.close();
+ }
+ }
+ } finally {
+ if (connection != null) {
+ connection.disconnect();
+ }
+ }
+ }
+
+ private static Map<String, Object> buildData(final String name, final String role, final String unit,
+ final String marker, final int concurrency,
+ final double mean, final double variance, final long n,
+ final double max, final double min, final double sum,
+ final double m2) {
+ final Map<String, Object> data = new HashMap<String, Object>();
+ data.put("name", name);
+ data.put("role", role);
+ data.put("unit", unit);
+ data.put("marker", marker);
+ data.put("concurrency", concurrency);
+ data.put("min", min);
+ data.put("mean", mean);
+ data.put("max", max);
+ data.put("variance", variance);
+ data.put("hits", n);
+ data.put("sum", sum);
+ data.put("m2", m2);
+ return data;
+ }
+
+ private static void assertCounter(final Counter counter, final int n, final double min, final double max,
+ final double mean, final double stdDev,
+ final double variance, final double sum) {
+ assertEquals(n, counter.getHits());
+ assertEquals(min, counter.getMin(), 0.);
+ assertEquals(max, counter.getMax(), 0.);
+ assertEquals(mean, counter.getMean(), 0.);
+ assertEquals(stdDev, counter.getStandardDeviation(), 0.0001);
+ assertEquals(variance, counter.getVariance(), 0.0001);
+ assertEquals(sum, counter.getSum(), 0.);
+ }
+}
Added: commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/store/counter/AggregatorTest.java
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/store/counter/AggregatorTest.java?rev=1534136&view=auto
==============================================================================
--- commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/store/counter/AggregatorTest.java (added)
+++ commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/store/counter/AggregatorTest.java Mon Oct 21 12:54:16 2013
@@ -0,0 +1,80 @@
+/*
+ * 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.commons.monitoring.collector.server.store.counter;
+
+import org.apache.commons.monitoring.Role;
+import org.apache.commons.monitoring.collector.server.math.M2AwareStatisticalSummary;
+import org.apache.commons.monitoring.counters.Counter;
+import org.apache.commons.monitoring.counters.Unit;
+import org.apache.commons.monitoring.repositories.Repository;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class AggregatorTest {
+ @Before @After
+ public void clear() {
+ Repository.INSTANCE.clear();
+ }
+
+ @Test
+ public void counterStore() {
+ final Counter.Key key = new Counter.Key(new Role("r", Unit.UNARY), "n");
+
+ // note: the input data are maybe not that consistent (min > max) but this test just checks computations
+ final CollectorCounterStore store = new InMemoryCollectorCounterStore();
+ store.update(key, "client1", new M2AwareStatisticalSummary(1, 2, 5, 0, 10, 6, 7), 4);
+ store.update(key, "client2", new M2AwareStatisticalSummary(2, 4, 8, 1, 15, 9, 5), 2);
+
+ assertEquals(2, store.markers().size());
+ assertTrue(store.markers().contains("client1"));
+ assertTrue(store.markers().contains("client2"));
+
+ final Counter counter1 = store.getCounters("client1").iterator().next();
+ assertEquals(4, counter1.getMaxConcurrency());
+ assertEquals(4, counter1.currentConcurrency().get());
+ assertEquals(5, counter1.getHits());
+ assertEquals(10., counter1.getMin(), 0);
+ assertEquals(0., counter1.getMax(), 0);
+ assertEquals(1.4142, counter1.getStandardDeviation(), 0.001);
+ assertEquals(2., counter1.getVariance(), 0);
+ assertEquals(6., counter1.getSum(), 0);
+
+ final Counter counter2 = store.getCounters("client2").iterator().next();
+ assertEquals(2, counter2.getMaxConcurrency());
+ assertEquals(2, counter2.currentConcurrency().get());
+ assertEquals(8, counter2.getHits());
+ assertEquals(15., counter2.getMin(), 0);
+ assertEquals(1., counter2.getMax(), 0);
+ assertEquals(2., counter2.getStandardDeviation(), 0.);
+ assertEquals(4., counter2.getVariance(), 0);
+ assertEquals(9., counter2.getSum(), 0);
+
+ final Counter aggregate = store.getOrCreateCounter(key);
+ assertEquals(6, aggregate.getMaxConcurrency());
+ assertEquals(6, aggregate.currentConcurrency().get());
+ assertEquals(13, aggregate.getHits());
+ assertEquals(10., aggregate.getMin(), 0);
+ assertEquals(1., aggregate.getMax(), 0);
+ assertEquals(1.12089, aggregate.getStandardDeviation(), 0.001);
+ assertEquals(1.2564, aggregate.getVariance(), 0.001);
+ assertEquals(15., aggregate.getSum(), 0);
+ }
+}
Added: commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/store/gauge/CollectorGaugeTest.java
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/store/gauge/CollectorGaugeTest.java?rev=1534136&view=auto
==============================================================================
--- commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/store/gauge/CollectorGaugeTest.java (added)
+++ commons/sandbox/monitoring/trunk/collector/src/test/java/org/apache/commons/monitoring/collector/server/store/gauge/CollectorGaugeTest.java Mon Oct 21 12:54:16 2013
@@ -0,0 +1,66 @@
+/*
+ * 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.commons.monitoring.collector.server.store.gauge;
+
+import org.apache.commons.monitoring.Role;
+import org.apache.commons.monitoring.counters.Unit;
+import org.apache.commons.monitoring.repositories.Repository;
+import org.apache.commons.monitoring.store.GaugeValuesRequest;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+public class CollectorGaugeTest {
+ @Before
+ @After
+ public void clear() {
+ Repository.INSTANCE.clear();
+ }
+
+ @Test
+ public void gaugeStore() {
+ final Role role = new Role("gauge", Unit.UNARY);
+
+ final CollectorGaugeStore store = new CollectorGaugeStore();
+ store.addToGauge(role, 1234, 5678, "client1");
+ store.addToGauge(role, 987, 654, "client2");
+
+ final GaugeValuesRequest request = new GaugeValuesRequest(0, Integer.MAX_VALUE, role);
+ final Map<Long, Double> result = store.getGaugeValues(request);
+ final Map<Long, Double> client1 = store.getGaugeValues(request, "client1");
+ final Map<Long, Double> client2 = store.getGaugeValues(request, "client2");
+
+ assertNotNull(result);
+ assertNotNull(client1);
+ assertNotNull(client2);
+
+ assertEquals(2, result.size());
+ assertEquals(654, result.get(987L), 0);
+ assertEquals(5678, result.get(1234L), 0);
+
+ assertEquals(1, client1.size());
+ assertEquals(5678, client1.get(1234L), 0);
+
+ assertEquals(1, client2.size());
+ assertEquals(654, client2.get(987L), 0);
+ }
+}
Copied: commons/sandbox/monitoring/trunk/collector/src/test/resources/commons-monitoring.properties (from r1534028, commons/sandbox/monitoring/trunk/cube/src/test/resources/commons-monitoring.properties)
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/collector/src/test/resources/commons-monitoring.properties?p2=commons/sandbox/monitoring/trunk/collector/src/test/resources/commons-monitoring.properties&p1=commons/sandbox/monitoring/trunk/cube/src/test/resources/commons-monitoring.properties&r1=1534028&r2=1534136&rev=1534136&view=diff
==============================================================================
--- commons/sandbox/monitoring/trunk/cube/src/test/resources/commons-monitoring.properties (original)
+++ commons/sandbox/monitoring/trunk/collector/src/test/resources/commons-monitoring.properties Mon Oct 21 12:54:16 2013
@@ -14,7 +14,7 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
-org.apache.commons.monitoring.store.DataStoreFactory = org.apache.commons.monitoring.cube.CubeDataStoreFactory
-org.apache.commons.monitoring.cube.period = 100
-org.apache.commons.monitoring.cube.CubeBuilder.collector = http://localhost:1234/1.0/event/put
+org.apache.commons.monitoring.store.DataStoreFactory = org.apache.commons.monitoring.collector.server.store.CollectorDataStoreFactory
+org.apache.commons.monitoring.cube.CubeBuilder.collector = http://localhost:1234
+org.apache.commons.monitoring.cube.CubeBuilder.marker = local
Modified: commons/sandbox/monitoring/trunk/core/pom.xml
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/core/pom.xml?rev=1534136&r1=1534135&r2=1534136&view=diff
==============================================================================
--- commons/sandbox/monitoring/trunk/core/pom.xml (original)
+++ commons/sandbox/monitoring/trunk/core/pom.xml Mon Oct 21 12:54:16 2013
@@ -34,7 +34,6 @@
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-math3</artifactId>
- <version>3.2</version>
</dependency>
<dependency>
Modified: commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/configuration/Configuration.java
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/configuration/Configuration.java?rev=1534136&r1=1534135&r2=1534136&view=diff
==============================================================================
--- commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/configuration/Configuration.java (original)
+++ commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/configuration/Configuration.java Mon Oct 21 12:54:16 2013
@@ -17,6 +17,7 @@
package org.apache.commons.monitoring.configuration;
import org.apache.commons.monitoring.MonitoringException;
+import org.apache.commons.monitoring.store.GaugeDataStore;
import org.apache.commons.monitoring.util.ClassLoaders;
import java.io.File;
@@ -175,6 +176,10 @@ public final class Configuration {
return clazz.cast(instance);
}
+ public static void setSingletonInstance(final Class<?> clazz, final Object instance) {
+ SINGLETONS.put(clazz, instance);
+ }
+
public static <T> T getInstance(final Class<T> clazz) {
return clazz.cast(SINGLETONS.get(clazz));
}
Modified: commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/counters/Counter.java
URL: http://svn.apache.org/viewvc/commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/counters/Counter.java?rev=1534136&r1=1534135&r2=1534136&view=diff
==============================================================================
--- commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/counters/Counter.java (original)
+++ commons/sandbox/monitoring/trunk/core/src/main/java/org/apache/commons/monitoring/counters/Counter.java Mon Oct 21 12:54:16 2013
@@ -58,11 +58,7 @@ public interface Counter {
double getMean();
- double getGeometricMean();
-
- double getSumOfLogs();
-
- double getSumOfSquares();
+ double getSecondMoment(); // here for aggregation etc but not (yet?) a human metric so not in MetricData
public static class Key {
private final String name;