@Test public void consume_data() throws Exception { when(request.getMethod()).thenReturn("POST"); when(request.getCharacterEncoding()).thenReturn("UTF-8"); when(request.getReader()).thenReturn(new BufferedReader(new StringReader(json))); when(response.getWriter()).thenReturn(new PrintWriter(new StringWriter())); Collection<MetricGroup> groups = GroupGenerator.deref(collectd.getGroups(threadpool, new CompletableFuture<>())); assertTrue(groups.isEmpty()); collectd_acceptor.service(request, response); groups = GroupGenerator.deref(collectd.getGroups(threadpool, new CompletableFuture<>())); Map<GroupName, Metric[]> group_map = groups.stream().map(x -> (MetricGroup) x).collect(Collectors.toMap(MetricGroup::getName, MetricGroup::getMetrics)); assertThat(group_map, allOf(hasKey(UPTIME_NAME), hasKey(DOWNTIME_NAME), hasKey(BASE_NAME))); assertEquals(UPTIME_METRIC, group_map.get(UPTIME_NAME)[0].getName()); assertEquals(MetricValue.fromIntValue(17), group_map.get(UPTIME_NAME)[0].getValue()); assertEquals(DOWNTIME_METRIC, group_map.get(DOWNTIME_NAME)[0].getName()); assertEquals(MetricValue.fromIntValue(-17), group_map.get(DOWNTIME_NAME)[0].getValue()); assertEquals(UP_METRIC, group_map.get(BASE_NAME)[0].getName()); assertEquals(MetricValue.TRUE, group_map.get(BASE_NAME)[0].getValue()); groups = GroupGenerator.deref(collectd.getGroups(threadpool, new CompletableFuture<>())); group_map = groups.stream().map(x -> (MetricGroup) x).collect(Collectors.toMap(MetricGroup::getName, MetricGroup::getMetrics)); assertEquals(group_map.keySet(), singleton(BASE_NAME)); assertEquals(UP_METRIC, group_map.get(BASE_NAME)[0].getName()); assertEquals(MetricValue.FALSE, group_map.get(BASE_NAME)[0].getValue()); verify(response, times(1)).setStatus(200); }
@Override public Collection<? extends MetricGroup> getGroups(CompletableFuture<TimeoutObject> timeout) { final DateTime now = new DateTime(DateTimeZone.UTC); final DateTime drop = now.minus(DROP_DURATION); Lock lck = messagesGuard.writeLock(); lck.lock(); try { final Stream<MetricGroup> msg_stream = messages.values().stream() .map(cm -> cm.toMetricGroup(basePath)) .flatMap(mg -> Arrays.stream(mg.getMetrics()).map(m -> SimpleMapEntry.create(mg.getName(), m))) .collect(Collectors.groupingBy(Map.Entry::getKey, Collectors.mapping(Map.Entry::getValue, Collectors.toList()))) .entrySet() .stream() .map((Map.Entry<GroupName, List<Metric>> entry) -> new SimpleMetricGroup(entry.getKey(), entry.getValue())); final Set<String> up_hosts = messages.values().stream() .map(msg -> msg.host) .distinct() .collect(Collectors.toSet()); final Map<String, DateTime> down_hosts = knownHosts.entrySet().stream() .filter(entry -> entry.getValue().isAfter(drop)) .filter(entry -> !up_hosts.contains(entry.getKey())) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); final Stream<MetricGroup> up_hosts_stream = up_hosts.stream() .map(host -> up_down_host_(host, true)); final Stream<MetricGroup> down_hosts_stream = down_hosts.keySet().stream() .map(host -> up_down_host_(host, false)); knownHosts = Stream.concat(down_hosts.entrySet().stream(), up_hosts.stream().map(host -> SimpleMapEntry.create(host, now))) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); return Stream.of(msg_stream, up_hosts_stream, down_hosts_stream) .flatMap(Function.identity()) .collect(Collectors.toList()); } finally { messages.clear(); lck.unlock(); } }
CollectdPushCollector extends SynchronousGroupGenerator { @Override public Collection<? extends MetricGroup> getGroups(CompletableFuture<TimeoutObject> timeout) { final DateTime now = new DateTime(DateTimeZone.UTC); final DateTime drop = now.minus(DROP_DURATION); Lock lck = messagesGuard.writeLock(); lck.lock(); try { final Stream<MetricGroup> msg_stream = messages.values().stream() .map(cm -> cm.toMetricGroup(basePath)) .flatMap(mg -> Arrays.stream(mg.getMetrics()).map(m -> SimpleMapEntry.create(mg.getName(), m))) .collect(Collectors.groupingBy(Map.Entry::getKey, Collectors.mapping(Map.Entry::getValue, Collectors.toList()))) .entrySet() .stream() .map((Map.Entry<GroupName, List<Metric>> entry) -> new SimpleMetricGroup(entry.getKey(), entry.getValue())); final Set<String> up_hosts = messages.values().stream() .map(msg -> msg.host) .distinct() .collect(Collectors.toSet()); final Map<String, DateTime> down_hosts = knownHosts.entrySet().stream() .filter(entry -> entry.getValue().isAfter(drop)) .filter(entry -> !up_hosts.contains(entry.getKey())) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); final Stream<MetricGroup> up_hosts_stream = up_hosts.stream() .map(host -> up_down_host_(host, true)); final Stream<MetricGroup> down_hosts_stream = down_hosts.keySet().stream() .map(host -> up_down_host_(host, false)); knownHosts = Stream.concat(down_hosts.entrySet().stream(), up_hosts.stream().map(host -> SimpleMapEntry.create(host, now))) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); return Stream.of(msg_stream, up_hosts_stream, down_hosts_stream) .flatMap(Function.identity()) .collect(Collectors.toList()); } finally { messages.clear(); lck.unlock(); } } }
CollectdPushCollector extends SynchronousGroupGenerator { @Override public Collection<? extends MetricGroup> getGroups(CompletableFuture<TimeoutObject> timeout) { final DateTime now = new DateTime(DateTimeZone.UTC); final DateTime drop = now.minus(DROP_DURATION); Lock lck = messagesGuard.writeLock(); lck.lock(); try { final Stream<MetricGroup> msg_stream = messages.values().stream() .map(cm -> cm.toMetricGroup(basePath)) .flatMap(mg -> Arrays.stream(mg.getMetrics()).map(m -> SimpleMapEntry.create(mg.getName(), m))) .collect(Collectors.groupingBy(Map.Entry::getKey, Collectors.mapping(Map.Entry::getValue, Collectors.toList()))) .entrySet() .stream() .map((Map.Entry<GroupName, List<Metric>> entry) -> new SimpleMetricGroup(entry.getKey(), entry.getValue())); final Set<String> up_hosts = messages.values().stream() .map(msg -> msg.host) .distinct() .collect(Collectors.toSet()); final Map<String, DateTime> down_hosts = knownHosts.entrySet().stream() .filter(entry -> entry.getValue().isAfter(drop)) .filter(entry -> !up_hosts.contains(entry.getKey())) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); final Stream<MetricGroup> up_hosts_stream = up_hosts.stream() .map(host -> up_down_host_(host, true)); final Stream<MetricGroup> down_hosts_stream = down_hosts.keySet().stream() .map(host -> up_down_host_(host, false)); knownHosts = Stream.concat(down_hosts.entrySet().stream(), up_hosts.stream().map(host -> SimpleMapEntry.create(host, now))) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); return Stream.of(msg_stream, up_hosts_stream, down_hosts_stream) .flatMap(Function.identity()) .collect(Collectors.toList()); } finally { messages.clear(); lck.unlock(); } } CollectdPushCollector(@NonNull BiConsumer<String, HttpServlet> er, @NonNull SimpleGroupPath base_path, @NonNull String name); }
CollectdPushCollector extends SynchronousGroupGenerator { @Override public Collection<? extends MetricGroup> getGroups(CompletableFuture<TimeoutObject> timeout) { final DateTime now = new DateTime(DateTimeZone.UTC); final DateTime drop = now.minus(DROP_DURATION); Lock lck = messagesGuard.writeLock(); lck.lock(); try { final Stream<MetricGroup> msg_stream = messages.values().stream() .map(cm -> cm.toMetricGroup(basePath)) .flatMap(mg -> Arrays.stream(mg.getMetrics()).map(m -> SimpleMapEntry.create(mg.getName(), m))) .collect(Collectors.groupingBy(Map.Entry::getKey, Collectors.mapping(Map.Entry::getValue, Collectors.toList()))) .entrySet() .stream() .map((Map.Entry<GroupName, List<Metric>> entry) -> new SimpleMetricGroup(entry.getKey(), entry.getValue())); final Set<String> up_hosts = messages.values().stream() .map(msg -> msg.host) .distinct() .collect(Collectors.toSet()); final Map<String, DateTime> down_hosts = knownHosts.entrySet().stream() .filter(entry -> entry.getValue().isAfter(drop)) .filter(entry -> !up_hosts.contains(entry.getKey())) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); final Stream<MetricGroup> up_hosts_stream = up_hosts.stream() .map(host -> up_down_host_(host, true)); final Stream<MetricGroup> down_hosts_stream = down_hosts.keySet().stream() .map(host -> up_down_host_(host, false)); knownHosts = Stream.concat(down_hosts.entrySet().stream(), up_hosts.stream().map(host -> SimpleMapEntry.create(host, now))) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); return Stream.of(msg_stream, up_hosts_stream, down_hosts_stream) .flatMap(Function.identity()) .collect(Collectors.toList()); } finally { messages.clear(); lck.unlock(); } } CollectdPushCollector(@NonNull BiConsumer<String, HttpServlet> er, @NonNull SimpleGroupPath base_path, @NonNull String name); @Override Collection<? extends MetricGroup> getGroups(CompletableFuture<TimeoutObject> timeout); }
CollectdPushCollector extends SynchronousGroupGenerator { @Override public Collection<? extends MetricGroup> getGroups(CompletableFuture<TimeoutObject> timeout) { final DateTime now = new DateTime(DateTimeZone.UTC); final DateTime drop = now.minus(DROP_DURATION); Lock lck = messagesGuard.writeLock(); lck.lock(); try { final Stream<MetricGroup> msg_stream = messages.values().stream() .map(cm -> cm.toMetricGroup(basePath)) .flatMap(mg -> Arrays.stream(mg.getMetrics()).map(m -> SimpleMapEntry.create(mg.getName(), m))) .collect(Collectors.groupingBy(Map.Entry::getKey, Collectors.mapping(Map.Entry::getValue, Collectors.toList()))) .entrySet() .stream() .map((Map.Entry<GroupName, List<Metric>> entry) -> new SimpleMetricGroup(entry.getKey(), entry.getValue())); final Set<String> up_hosts = messages.values().stream() .map(msg -> msg.host) .distinct() .collect(Collectors.toSet()); final Map<String, DateTime> down_hosts = knownHosts.entrySet().stream() .filter(entry -> entry.getValue().isAfter(drop)) .filter(entry -> !up_hosts.contains(entry.getKey())) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); final Stream<MetricGroup> up_hosts_stream = up_hosts.stream() .map(host -> up_down_host_(host, true)); final Stream<MetricGroup> down_hosts_stream = down_hosts.keySet().stream() .map(host -> up_down_host_(host, false)); knownHosts = Stream.concat(down_hosts.entrySet().stream(), up_hosts.stream().map(host -> SimpleMapEntry.create(host, now))) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); return Stream.of(msg_stream, up_hosts_stream, down_hosts_stream) .flatMap(Function.identity()) .collect(Collectors.toList()); } finally { messages.clear(); lck.unlock(); } } CollectdPushCollector(@NonNull BiConsumer<String, HttpServlet> er, @NonNull SimpleGroupPath base_path, @NonNull String name); @Override Collection<? extends MetricGroup> getGroups(CompletableFuture<TimeoutObject> timeout); static final String API_ENDPOINT_BASE; static final Duration DROP_DURATION; }