@Test public void test() { Map<String, Long> accumulators = Maps.newHashMap(); accumulators.put("global.hosts.www.leboncoin.com", 1l); accumulators.put("global.nbPagesTotal", 1l); accumulators.put("job_1.hosts.www.leboncoin.com", 1l); accumulators.put("job_1.statuses.200", 5l); accumulators.put("node_1.hosts.www.leboncoin.com", 1l); accumulators.put("node_1.statuses.200", 5l); mongoMetricsRepository.sync(accumulators); }
@Override public void sync(Map<String, Long> accumulators) { LocalDateTime now = LocalDateTime.now(); LocalDateTime keytime = now.withMinute(0).withSecond(0).withNano(0); Stream<Pair<String, Pair<String, Long>>> map = accumulators.entrySet().stream().map(e -> { Iterable<String> results = splitter.split(e.getKey()); List<String> elts = Lists.newArrayList(results); return Pair.of(elts.get(0), Pair.of(elts.get(1), e.getValue())); }); Map<String, List<Pair<String, Long>>> byKey = map.collect(Collectors.groupingBy(e -> e.getLeft(), Collectors.mapping(e -> e.getRight(), Collectors.toList()))); List<? extends WriteModel<Document>> requests = byKey.entrySet().stream().map(e -> { Document updates = new Document(); e.getValue().stream().forEach(i -> { Iterable<String> results = splitter.split(i.getKey()); List<String> elts = Lists.newArrayList(results); if (elts.size() > 1) { updates.put(elts.get(0) + "." + JsonBsonCodec.toBson(elts.get(1)), i.getValue()); } else { updates.put(i.getKey(), i.getValue()); } }); return new UpdateOneModel<Document>(Filters.eq("_id", e.getKey()), new Document("$inc", updates), new UpdateOptions().upsert(true)); }).collect(Collectors.toList()); counters.bulkWrite(requests); requests = byKey .entrySet() .stream() .map(e -> { List<UpdateOneModel<Document>> tsUpdates = Lists.newArrayList(); e.getValue() .stream() .forEach( i -> { Iterable<String> results = splitter.split(i.getKey()); List<String> elts = Lists.newArrayList(results); if (elts.size() == 1 && e.getKey().equalsIgnoreCase(MetricKeys.global())) { tsUpdates.add(new UpdateOneModel<Document>(Filters.and( Filters.eq("type", e.getKey() + MetricKeys.METRIC_DELIM + i.getKey()), Filters.eq("timestamp_hour", keytime)), new Document("$inc", new Document("values." + Integer.toString(now.getMinute()), i.getValue())), new UpdateOptions().upsert(true))); } }); return tsUpdates; }).flatMap(list -> list.stream()).collect(Collectors.toList()); timeseries.bulkWrite(requests); }
MongoMetricsRepository implements MetricsRepository { @Override public void sync(Map<String, Long> accumulators) { LocalDateTime now = LocalDateTime.now(); LocalDateTime keytime = now.withMinute(0).withSecond(0).withNano(0); Stream<Pair<String, Pair<String, Long>>> map = accumulators.entrySet().stream().map(e -> { Iterable<String> results = splitter.split(e.getKey()); List<String> elts = Lists.newArrayList(results); return Pair.of(elts.get(0), Pair.of(elts.get(1), e.getValue())); }); Map<String, List<Pair<String, Long>>> byKey = map.collect(Collectors.groupingBy(e -> e.getLeft(), Collectors.mapping(e -> e.getRight(), Collectors.toList()))); List<? extends WriteModel<Document>> requests = byKey.entrySet().stream().map(e -> { Document updates = new Document(); e.getValue().stream().forEach(i -> { Iterable<String> results = splitter.split(i.getKey()); List<String> elts = Lists.newArrayList(results); if (elts.size() > 1) { updates.put(elts.get(0) + "." + JsonBsonCodec.toBson(elts.get(1)), i.getValue()); } else { updates.put(i.getKey(), i.getValue()); } }); return new UpdateOneModel<Document>(Filters.eq("_id", e.getKey()), new Document("$inc", updates), new UpdateOptions().upsert(true)); }).collect(Collectors.toList()); counters.bulkWrite(requests); requests = byKey .entrySet() .stream() .map(e -> { List<UpdateOneModel<Document>> tsUpdates = Lists.newArrayList(); e.getValue() .stream() .forEach( i -> { Iterable<String> results = splitter.split(i.getKey()); List<String> elts = Lists.newArrayList(results); if (elts.size() == 1 && e.getKey().equalsIgnoreCase(MetricKeys.global())) { tsUpdates.add(new UpdateOneModel<Document>(Filters.and( Filters.eq("type", e.getKey() + MetricKeys.METRIC_DELIM + i.getKey()), Filters.eq("timestamp_hour", keytime)), new Document("$inc", new Document("values." + Integer.toString(now.getMinute()), i.getValue())), new UpdateOptions().upsert(true))); } }); return tsUpdates; }).flatMap(list -> list.stream()).collect(Collectors.toList()); timeseries.bulkWrite(requests); } }
MongoMetricsRepository implements MetricsRepository { @Override public void sync(Map<String, Long> accumulators) { LocalDateTime now = LocalDateTime.now(); LocalDateTime keytime = now.withMinute(0).withSecond(0).withNano(0); Stream<Pair<String, Pair<String, Long>>> map = accumulators.entrySet().stream().map(e -> { Iterable<String> results = splitter.split(e.getKey()); List<String> elts = Lists.newArrayList(results); return Pair.of(elts.get(0), Pair.of(elts.get(1), e.getValue())); }); Map<String, List<Pair<String, Long>>> byKey = map.collect(Collectors.groupingBy(e -> e.getLeft(), Collectors.mapping(e -> e.getRight(), Collectors.toList()))); List<? extends WriteModel<Document>> requests = byKey.entrySet().stream().map(e -> { Document updates = new Document(); e.getValue().stream().forEach(i -> { Iterable<String> results = splitter.split(i.getKey()); List<String> elts = Lists.newArrayList(results); if (elts.size() > 1) { updates.put(elts.get(0) + "." + JsonBsonCodec.toBson(elts.get(1)), i.getValue()); } else { updates.put(i.getKey(), i.getValue()); } }); return new UpdateOneModel<Document>(Filters.eq("_id", e.getKey()), new Document("$inc", updates), new UpdateOptions().upsert(true)); }).collect(Collectors.toList()); counters.bulkWrite(requests); requests = byKey .entrySet() .stream() .map(e -> { List<UpdateOneModel<Document>> tsUpdates = Lists.newArrayList(); e.getValue() .stream() .forEach( i -> { Iterable<String> results = splitter.split(i.getKey()); List<String> elts = Lists.newArrayList(results); if (elts.size() == 1 && e.getKey().equalsIgnoreCase(MetricKeys.global())) { tsUpdates.add(new UpdateOneModel<Document>(Filters.and( Filters.eq("type", e.getKey() + MetricKeys.METRIC_DELIM + i.getKey()), Filters.eq("timestamp_hour", keytime)), new Document("$inc", new Document("values." + Integer.toString(now.getMinute()), i.getValue())), new UpdateOptions().upsert(true))); } }); return tsUpdates; }).flatMap(list -> list.stream()).collect(Collectors.toList()); timeseries.bulkWrite(requests); }  }
MongoMetricsRepository implements MetricsRepository { @Override public void sync(Map<String, Long> accumulators) { LocalDateTime now = LocalDateTime.now(); LocalDateTime keytime = now.withMinute(0).withSecond(0).withNano(0); Stream<Pair<String, Pair<String, Long>>> map = accumulators.entrySet().stream().map(e -> { Iterable<String> results = splitter.split(e.getKey()); List<String> elts = Lists.newArrayList(results); return Pair.of(elts.get(0), Pair.of(elts.get(1), e.getValue())); }); Map<String, List<Pair<String, Long>>> byKey = map.collect(Collectors.groupingBy(e -> e.getLeft(), Collectors.mapping(e -> e.getRight(), Collectors.toList()))); List<? extends WriteModel<Document>> requests = byKey.entrySet().stream().map(e -> { Document updates = new Document(); e.getValue().stream().forEach(i -> { Iterable<String> results = splitter.split(i.getKey()); List<String> elts = Lists.newArrayList(results); if (elts.size() > 1) { updates.put(elts.get(0) + "." + JsonBsonCodec.toBson(elts.get(1)), i.getValue()); } else { updates.put(i.getKey(), i.getValue()); } }); return new UpdateOneModel<Document>(Filters.eq("_id", e.getKey()), new Document("$inc", updates), new UpdateOptions().upsert(true)); }).collect(Collectors.toList()); counters.bulkWrite(requests); requests = byKey .entrySet() .stream() .map(e -> { List<UpdateOneModel<Document>> tsUpdates = Lists.newArrayList(); e.getValue() .stream() .forEach( i -> { Iterable<String> results = splitter.split(i.getKey()); List<String> elts = Lists.newArrayList(results); if (elts.size() == 1 && e.getKey().equalsIgnoreCase(MetricKeys.global())) { tsUpdates.add(new UpdateOneModel<Document>(Filters.and( Filters.eq("type", e.getKey() + MetricKeys.METRIC_DELIM + i.getKey()), Filters.eq("timestamp_hour", keytime)), new Document("$inc", new Document("values." + Integer.toString(now.getMinute()), i.getValue())), new UpdateOptions().upsert(true))); } }); return tsUpdates; }).flatMap(list -> list.stream()).collect(Collectors.toList()); timeseries.bulkWrite(requests); }  @PostConstruct void init(); void checkFilled(); @Override void sync(Map<String, Long> accumulators); @Scheduled(cron = "0 0 * * * *") void prepareNextMinutes(); void prepareMinutes(LocalDateTime keytime); @Override NodeMetrics node(String nodeId); @Override GlobalMetrics global(); @Override JobMetrics job(long jobId); @Override void delete(long jobId); @Override Timeserie serie(String name); }
MongoMetricsRepository implements MetricsRepository { @Override public void sync(Map<String, Long> accumulators) { LocalDateTime now = LocalDateTime.now(); LocalDateTime keytime = now.withMinute(0).withSecond(0).withNano(0); Stream<Pair<String, Pair<String, Long>>> map = accumulators.entrySet().stream().map(e -> { Iterable<String> results = splitter.split(e.getKey()); List<String> elts = Lists.newArrayList(results); return Pair.of(elts.get(0), Pair.of(elts.get(1), e.getValue())); }); Map<String, List<Pair<String, Long>>> byKey = map.collect(Collectors.groupingBy(e -> e.getLeft(), Collectors.mapping(e -> e.getRight(), Collectors.toList()))); List<? extends WriteModel<Document>> requests = byKey.entrySet().stream().map(e -> { Document updates = new Document(); e.getValue().stream().forEach(i -> { Iterable<String> results = splitter.split(i.getKey()); List<String> elts = Lists.newArrayList(results); if (elts.size() > 1) { updates.put(elts.get(0) + "." + JsonBsonCodec.toBson(elts.get(1)), i.getValue()); } else { updates.put(i.getKey(), i.getValue()); } }); return new UpdateOneModel<Document>(Filters.eq("_id", e.getKey()), new Document("$inc", updates), new UpdateOptions().upsert(true)); }).collect(Collectors.toList()); counters.bulkWrite(requests); requests = byKey .entrySet() .stream() .map(e -> { List<UpdateOneModel<Document>> tsUpdates = Lists.newArrayList(); e.getValue() .stream() .forEach( i -> { Iterable<String> results = splitter.split(i.getKey()); List<String> elts = Lists.newArrayList(results); if (elts.size() == 1 && e.getKey().equalsIgnoreCase(MetricKeys.global())) { tsUpdates.add(new UpdateOneModel<Document>(Filters.and( Filters.eq("type", e.getKey() + MetricKeys.METRIC_DELIM + i.getKey()), Filters.eq("timestamp_hour", keytime)), new Document("$inc", new Document("values." + Integer.toString(now.getMinute()), i.getValue())), new UpdateOptions().upsert(true))); } }); return tsUpdates; }).flatMap(list -> list.stream()).collect(Collectors.toList()); timeseries.bulkWrite(requests); }  @PostConstruct void init(); void checkFilled(); @Override void sync(Map<String, Long> accumulators); @Scheduled(cron = "0 0 * * * *") void prepareNextMinutes(); void prepareMinutes(LocalDateTime keytime); @Override NodeMetrics node(String nodeId); @Override GlobalMetrics global(); @Override JobMetrics job(long jobId); @Override void delete(long jobId); @Override Timeserie serie(String name);  }