HadoopTimelineMetricsSink extends AbstractTimelineMetricsSink implements MetricsSink, Closeable { @Override public void putMetrics(MetricsRecord record) { try { String recordName = record.name(); String contextName = record.context(); StringBuilder sb = new StringBuilder(); boolean skipAggregation = false; if (contextName.startsWith("ipc.")) { String portNumber = contextName.replaceFirst("ipc.", ""); if (rpcPortSuffixes.containsKey(portNumber)) { contextName = "ipc." + rpcPortSuffixes.get(portNumber); } } sb.append(contextName); sb.append('.'); if (record.tags() != null) { for (MetricsTag tag : record.tags()) { if (StringUtils.isNotEmpty(tag.name()) && tag.name().equals("skipAggregation")) { skipAggregation = String.valueOf(true).equals(tag.value()); } if (contextName.equals("jvm") && tag.info().name().equalsIgnoreCase("processName") && (tag.value().equals("RegionServer") || tag.value().equals("Master"))) { sb.append(tag.value()); sb.append('.'); } } } sb.append(recordName); appendPrefix(record, sb); sb.append('.'); if (!rpcPortSuffixes.isEmpty() && contextName.contains("rpc")) { if (record.tags() != null) { for (MetricsTag tag : record.tags()) { if (tag.info().name().equalsIgnoreCase("port") && rpcPortSuffixes.keySet().contains(tag.value())) { sb.append(rpcPortSuffixes.get(tag.value())); sb.append('.'); } } } } if (record.context().equals("container")) { emitContainerMetrics(record); return; } int sbBaseLen = sb.length(); List<TimelineMetric> metricList = new ArrayList<TimelineMetric>(); Map<String, String> metadata = null; if (skipAggregation) { metadata = Collections.singletonMap("skipAggregation", "true"); } long startTime = record.timestamp(); for (AbstractMetric metric : record.metrics()) { sb.append(metric.name()); String name = sb.toString(); Number value = metric.value(); TimelineMetric timelineMetric = new TimelineMetric(); timelineMetric.setMetricName(name); timelineMetric.setHostName(hostName); timelineMetric.setAppId(serviceName); timelineMetric.setStartTime(startTime); timelineMetric.setType(metric.type() != null ? metric.type().name() : null); timelineMetric.getMetricValues().put(startTime, value.doubleValue()); if (metadata != null) { timelineMetric.setMetadata(metadata); } boolean isCounter = MetricType.COUNTER == metric.type(); metricsCache.putTimelineMetric(timelineMetric, isCounter); TimelineMetric cachedMetric = metricsCache.getTimelineMetric(name); if (cachedMetric != null) { metricList.add(cachedMetric); } sb.setLength(sbBaseLen); } TimelineMetrics timelineMetrics = new TimelineMetrics(); timelineMetrics.setMetrics(metricList); if (!metricList.isEmpty()) { emitMetrics(timelineMetrics); } } catch (UnableToConnectException uce) { LOG.warn("Unable to send metrics to collector by address:" + uce.getConnectUrl()); } }  @Override void init(SubsetConfiguration conf); @Override void putMetrics(MetricsRecord record); @InterfaceAudience.Private void appendPrefix(MetricsRecord record, StringBuilder sb); @Override void flush(); @Override void close(); static final String WS_V1_CONTAINER_METRICS; }
@Test @PrepareForTest({URL.class, OutputStream.class, AbstractTimelineMetricsSink.class, HttpURLConnection.class}) public void testPutMetrics() throws Exception { HadoopTimelineMetricsSink sink = new HadoopTimelineMetricsSink(); HttpURLConnection connection = PowerMock.createNiceMock(HttpURLConnection.class); URL url = PowerMock.createNiceMock(URL.class); InputStream is = IOUtils.toInputStream(gson.toJson(Collections.singletonList("localhost"))); expectNew(URL.class, anyString()).andReturn(url).anyTimes(); expect(url.openConnection()).andReturn(connection).anyTimes(); expect(connection.getInputStream()).andReturn(is).anyTimes(); expect(connection.getResponseCode()).andReturn(200).anyTimes(); OutputStream os = PowerMock.createNiceMock(OutputStream.class); expect(connection.getOutputStream()).andReturn(os).anyTimes(); SubsetConfiguration conf = createNiceMock(SubsetConfiguration.class); expect(conf.getString("slave.host.name")).andReturn("localhost").anyTimes(); expect(conf.getParent()).andReturn(null).anyTimes(); expect(conf.getPrefix()).andReturn("service").anyTimes(); expect(conf.getStringArray(eq(COLLECTOR_HOSTS_PROPERTY))).andReturn(new String[]{"localhost"," localhost2"}).anyTimes(); expect(conf.getString(eq("serviceName-prefix"), eq(""))).andReturn("").anyTimes(); expect(conf.getString(eq(COLLECTOR_PROTOCOL), eq("http"))).andReturn("http").anyTimes(); expect(conf.getString(eq(COLLECTOR_PORT), eq("6188"))).andReturn("6188").anyTimes(); expect(conf.getInt(eq(MAX_METRIC_ROW_CACHE_SIZE), anyInt())).andReturn(10).anyTimes(); expect(conf.getInt(eq(METRICS_SEND_INTERVAL), anyInt())).andReturn(1000).anyTimes(); conf.setListDelimiter(eq(',')); expectLastCall().anyTimes(); expect(conf.getKeys()).andReturn(new Iterator() { @Override public boolean hasNext() { return false; } @Override public Object next() { return null; } @Override public void remove() { } }).once(); AbstractMetric metric = createNiceMock(AbstractMetric.class); expect(metric.name()).andReturn("metricName").anyTimes(); expect(metric.value()).andReturn(9.5687).anyTimes(); expect(metric.type()).andReturn(MetricType.COUNTER).anyTimes(); MetricsRecord record = createNiceMock(MetricsRecord.class); expect(record.name()).andReturn("testName").anyTimes(); expect(record.context()).andReturn("testContext").anyTimes(); expect(record.timestamp()).andAnswer(new IAnswer<Long>() { @Override public Long answer() throws Throwable { return System.currentTimeMillis(); } }).anyTimes(); expect(record.metrics()).andReturn(Arrays.asList(metric)).anyTimes(); replay(conf, record, metric); replayAll(); sink.init(conf); sink.putMetrics(record); Thread.sleep(1500L); sink.putMetrics(record); verifyAll(); }