@Test public void testAmbariFormat() throws IOException, InitializationException { final Map<PropertyDescriptor, String> properties = new HashMap<>(); properties.put(SiteToSiteMetricsReportingTask.FORMAT, SiteToSiteMetricsReportingTask.AMBARI_FORMAT.getValue()); MockSiteToSiteMetricsReportingTask task = initTask(properties); task.onTrigger(context); assertEquals(1, task.dataSent.size()); final String msg = new String(task.dataSent.get(0), StandardCharsets.UTF_8); JsonReader jsonReader = Json.createReader(new ByteArrayInputStream(msg.getBytes())); JsonArray array = jsonReader.readObject().getJsonArray("metrics"); for(int i = 0; i < array.size(); i++) { JsonObject object = array.getJsonObject(i); assertEquals("nifi", object.getString("appid")); assertEquals("1234", object.getString("instanceid")); if(object.getString("metricname").equals("FlowFilesQueued")) { for(Entry<String, JsonValue> kv : object.getJsonObject("metrics").entrySet()) { assertEquals("\"100\"", kv.getValue().toString()); } return; } } fail(); }
@Override public void onTrigger(final ReportingContext context) { final boolean isClustered = context.isClustered(); final String nodeId = context.getClusterNodeIdentifier(); if (nodeId == null && isClustered) { getLogger().debug("This instance of NiFi is configured for clustering, but the Cluster Node Identifier is not yet available. " + "Will wait for Node Identifier to be established."); return; } final VirtualMachineMetrics virtualMachineMetrics = VirtualMachineMetrics.getInstance(); final Map<String, ?> config = Collections.emptyMap(); final JsonBuilderFactory factory = Json.createBuilderFactory(config); final String applicationId = context.getProperty(APPLICATION_ID).evaluateAttributeExpressions().getValue(); final String hostname = context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue(); final ProcessGroupStatus status = context.getEventAccess().getControllerStatus(); if(status != null) { final Map<String,String> statusMetrics = metricsService.getMetrics(status, false); final Map<String,String> jvmMetrics = metricsService.getMetrics(virtualMachineMetrics); final MetricsBuilder metricsBuilder = new MetricsBuilder(factory); final OperatingSystemMXBean os = ManagementFactory.getOperatingSystemMXBean(); final double systemLoad = os.getSystemLoadAverage(); byte[] data; final Map<String, String> attributes = new HashMap<>(); if(context.getProperty(FORMAT).getValue().equals(AMBARI_FORMAT.getValue())) { final JsonObject metricsObject = metricsBuilder .applicationId(applicationId) .instanceId(status.getId()) .hostname(hostname) .timestamp(System.currentTimeMillis()) .addAllMetrics(statusMetrics) .addAllMetrics(jvmMetrics) .metric(MetricNames.CORES, String.valueOf(os.getAvailableProcessors())) .metric(MetricNames.LOAD1MN, String.valueOf(systemLoad >= 0 ? systemLoad : -1)) .build(); data = metricsObject.toString().getBytes(StandardCharsets.UTF_8); attributes.put(CoreAttributes.MIME_TYPE.key(), "application/json"); } else { final JsonObject metricsObject = metricsService.getMetrics(factory, status, virtualMachineMetrics, applicationId, status.getId(), hostname, System.currentTimeMillis(), os.getAvailableProcessors(), systemLoad >= 0 ? systemLoad : -1); data = getData(context, new ByteArrayInputStream(metricsObject.toString().getBytes(StandardCharsets.UTF_8)), attributes); } try { long start = System.nanoTime(); final Transaction transaction = getClient().createTransaction(TransferDirection.SEND); if (transaction == null) { getLogger().debug("All destination nodes are penalized; will attempt to send data later"); return; } final String transactionId = UUID.randomUUID().toString(); attributes.put("reporting.task.transaction.id", transactionId); attributes.put("reporting.task.name", getName()); attributes.put("reporting.task.uuid", getIdentifier()); attributes.put("reporting.task.type", this.getClass().getSimpleName()); transaction.send(data, attributes); transaction.confirm(); transaction.complete(); final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); getLogger().info("Successfully sent metrics to destination in {}ms; Transaction ID = {}", new Object[]{transferMillis, transactionId}); } catch (final Exception e) { throw new ProcessException("Failed to send metrics to destination due to:" + e.getMessage(), e); } } else { getLogger().error("No process group status to retrieve metrics"); } }
SiteToSiteMetricsReportingTask extends AbstractSiteToSiteReportingTask { @Override public void onTrigger(final ReportingContext context) { final boolean isClustered = context.isClustered(); final String nodeId = context.getClusterNodeIdentifier(); if (nodeId == null && isClustered) { getLogger().debug("This instance of NiFi is configured for clustering, but the Cluster Node Identifier is not yet available. " + "Will wait for Node Identifier to be established."); return; } final VirtualMachineMetrics virtualMachineMetrics = VirtualMachineMetrics.getInstance(); final Map<String, ?> config = Collections.emptyMap(); final JsonBuilderFactory factory = Json.createBuilderFactory(config); final String applicationId = context.getProperty(APPLICATION_ID).evaluateAttributeExpressions().getValue(); final String hostname = context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue(); final ProcessGroupStatus status = context.getEventAccess().getControllerStatus(); if(status != null) { final Map<String,String> statusMetrics = metricsService.getMetrics(status, false); final Map<String,String> jvmMetrics = metricsService.getMetrics(virtualMachineMetrics); final MetricsBuilder metricsBuilder = new MetricsBuilder(factory); final OperatingSystemMXBean os = ManagementFactory.getOperatingSystemMXBean(); final double systemLoad = os.getSystemLoadAverage(); byte[] data; final Map<String, String> attributes = new HashMap<>(); if(context.getProperty(FORMAT).getValue().equals(AMBARI_FORMAT.getValue())) { final JsonObject metricsObject = metricsBuilder .applicationId(applicationId) .instanceId(status.getId()) .hostname(hostname) .timestamp(System.currentTimeMillis()) .addAllMetrics(statusMetrics) .addAllMetrics(jvmMetrics) .metric(MetricNames.CORES, String.valueOf(os.getAvailableProcessors())) .metric(MetricNames.LOAD1MN, String.valueOf(systemLoad >= 0 ? systemLoad : -1)) .build(); data = metricsObject.toString().getBytes(StandardCharsets.UTF_8); attributes.put(CoreAttributes.MIME_TYPE.key(), "application/json"); } else { final JsonObject metricsObject = metricsService.getMetrics(factory, status, virtualMachineMetrics, applicationId, status.getId(), hostname, System.currentTimeMillis(), os.getAvailableProcessors(), systemLoad >= 0 ? systemLoad : -1); data = getData(context, new ByteArrayInputStream(metricsObject.toString().getBytes(StandardCharsets.UTF_8)), attributes); } try { long start = System.nanoTime(); final Transaction transaction = getClient().createTransaction(TransferDirection.SEND); if (transaction == null) { getLogger().debug("All destination nodes are penalized; will attempt to send data later"); return; } final String transactionId = UUID.randomUUID().toString(); attributes.put("reporting.task.transaction.id", transactionId); attributes.put("reporting.task.name", getName()); attributes.put("reporting.task.uuid", getIdentifier()); attributes.put("reporting.task.type", this.getClass().getSimpleName()); transaction.send(data, attributes); transaction.confirm(); transaction.complete(); final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); getLogger().info("Successfully sent metrics to destination in {}ms; Transaction ID = {}", new Object[]{transferMillis, transactionId}); } catch (final Exception e) { throw new ProcessException("Failed to send metrics to destination due to:" + e.getMessage(), e); } } else { getLogger().error("No process group status to retrieve metrics"); } } }
SiteToSiteMetricsReportingTask extends AbstractSiteToSiteReportingTask { @Override public void onTrigger(final ReportingContext context) { final boolean isClustered = context.isClustered(); final String nodeId = context.getClusterNodeIdentifier(); if (nodeId == null && isClustered) { getLogger().debug("This instance of NiFi is configured for clustering, but the Cluster Node Identifier is not yet available. " + "Will wait for Node Identifier to be established."); return; } final VirtualMachineMetrics virtualMachineMetrics = VirtualMachineMetrics.getInstance(); final Map<String, ?> config = Collections.emptyMap(); final JsonBuilderFactory factory = Json.createBuilderFactory(config); final String applicationId = context.getProperty(APPLICATION_ID).evaluateAttributeExpressions().getValue(); final String hostname = context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue(); final ProcessGroupStatus status = context.getEventAccess().getControllerStatus(); if(status != null) { final Map<String,String> statusMetrics = metricsService.getMetrics(status, false); final Map<String,String> jvmMetrics = metricsService.getMetrics(virtualMachineMetrics); final MetricsBuilder metricsBuilder = new MetricsBuilder(factory); final OperatingSystemMXBean os = ManagementFactory.getOperatingSystemMXBean(); final double systemLoad = os.getSystemLoadAverage(); byte[] data; final Map<String, String> attributes = new HashMap<>(); if(context.getProperty(FORMAT).getValue().equals(AMBARI_FORMAT.getValue())) { final JsonObject metricsObject = metricsBuilder .applicationId(applicationId) .instanceId(status.getId()) .hostname(hostname) .timestamp(System.currentTimeMillis()) .addAllMetrics(statusMetrics) .addAllMetrics(jvmMetrics) .metric(MetricNames.CORES, String.valueOf(os.getAvailableProcessors())) .metric(MetricNames.LOAD1MN, String.valueOf(systemLoad >= 0 ? systemLoad : -1)) .build(); data = metricsObject.toString().getBytes(StandardCharsets.UTF_8); attributes.put(CoreAttributes.MIME_TYPE.key(), "application/json"); } else { final JsonObject metricsObject = metricsService.getMetrics(factory, status, virtualMachineMetrics, applicationId, status.getId(), hostname, System.currentTimeMillis(), os.getAvailableProcessors(), systemLoad >= 0 ? systemLoad : -1); data = getData(context, new ByteArrayInputStream(metricsObject.toString().getBytes(StandardCharsets.UTF_8)), attributes); } try { long start = System.nanoTime(); final Transaction transaction = getClient().createTransaction(TransferDirection.SEND); if (transaction == null) { getLogger().debug("All destination nodes are penalized; will attempt to send data later"); return; } final String transactionId = UUID.randomUUID().toString(); attributes.put("reporting.task.transaction.id", transactionId); attributes.put("reporting.task.name", getName()); attributes.put("reporting.task.uuid", getIdentifier()); attributes.put("reporting.task.type", this.getClass().getSimpleName()); transaction.send(data, attributes); transaction.confirm(); transaction.complete(); final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); getLogger().info("Successfully sent metrics to destination in {}ms; Transaction ID = {}", new Object[]{transferMillis, transactionId}); } catch (final Exception e) { throw new ProcessException("Failed to send metrics to destination due to:" + e.getMessage(), e); } } else { getLogger().error("No process group status to retrieve metrics"); } } SiteToSiteMetricsReportingTask(); }
SiteToSiteMetricsReportingTask extends AbstractSiteToSiteReportingTask { @Override public void onTrigger(final ReportingContext context) { final boolean isClustered = context.isClustered(); final String nodeId = context.getClusterNodeIdentifier(); if (nodeId == null && isClustered) { getLogger().debug("This instance of NiFi is configured for clustering, but the Cluster Node Identifier is not yet available. " + "Will wait for Node Identifier to be established."); return; } final VirtualMachineMetrics virtualMachineMetrics = VirtualMachineMetrics.getInstance(); final Map<String, ?> config = Collections.emptyMap(); final JsonBuilderFactory factory = Json.createBuilderFactory(config); final String applicationId = context.getProperty(APPLICATION_ID).evaluateAttributeExpressions().getValue(); final String hostname = context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue(); final ProcessGroupStatus status = context.getEventAccess().getControllerStatus(); if(status != null) { final Map<String,String> statusMetrics = metricsService.getMetrics(status, false); final Map<String,String> jvmMetrics = metricsService.getMetrics(virtualMachineMetrics); final MetricsBuilder metricsBuilder = new MetricsBuilder(factory); final OperatingSystemMXBean os = ManagementFactory.getOperatingSystemMXBean(); final double systemLoad = os.getSystemLoadAverage(); byte[] data; final Map<String, String> attributes = new HashMap<>(); if(context.getProperty(FORMAT).getValue().equals(AMBARI_FORMAT.getValue())) { final JsonObject metricsObject = metricsBuilder .applicationId(applicationId) .instanceId(status.getId()) .hostname(hostname) .timestamp(System.currentTimeMillis()) .addAllMetrics(statusMetrics) .addAllMetrics(jvmMetrics) .metric(MetricNames.CORES, String.valueOf(os.getAvailableProcessors())) .metric(MetricNames.LOAD1MN, String.valueOf(systemLoad >= 0 ? systemLoad : -1)) .build(); data = metricsObject.toString().getBytes(StandardCharsets.UTF_8); attributes.put(CoreAttributes.MIME_TYPE.key(), "application/json"); } else { final JsonObject metricsObject = metricsService.getMetrics(factory, status, virtualMachineMetrics, applicationId, status.getId(), hostname, System.currentTimeMillis(), os.getAvailableProcessors(), systemLoad >= 0 ? systemLoad : -1); data = getData(context, new ByteArrayInputStream(metricsObject.toString().getBytes(StandardCharsets.UTF_8)), attributes); } try { long start = System.nanoTime(); final Transaction transaction = getClient().createTransaction(TransferDirection.SEND); if (transaction == null) { getLogger().debug("All destination nodes are penalized; will attempt to send data later"); return; } final String transactionId = UUID.randomUUID().toString(); attributes.put("reporting.task.transaction.id", transactionId); attributes.put("reporting.task.name", getName()); attributes.put("reporting.task.uuid", getIdentifier()); attributes.put("reporting.task.type", this.getClass().getSimpleName()); transaction.send(data, attributes); transaction.confirm(); transaction.complete(); final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); getLogger().info("Successfully sent metrics to destination in {}ms; Transaction ID = {}", new Object[]{transferMillis, transactionId}); } catch (final Exception e) { throw new ProcessException("Failed to send metrics to destination due to:" + e.getMessage(), e); } } else { getLogger().error("No process group status to retrieve metrics"); } } SiteToSiteMetricsReportingTask(); @Override void onTrigger(final ReportingContext context); }
SiteToSiteMetricsReportingTask extends AbstractSiteToSiteReportingTask { @Override public void onTrigger(final ReportingContext context) { final boolean isClustered = context.isClustered(); final String nodeId = context.getClusterNodeIdentifier(); if (nodeId == null && isClustered) { getLogger().debug("This instance of NiFi is configured for clustering, but the Cluster Node Identifier is not yet available. " + "Will wait for Node Identifier to be established."); return; } final VirtualMachineMetrics virtualMachineMetrics = VirtualMachineMetrics.getInstance(); final Map<String, ?> config = Collections.emptyMap(); final JsonBuilderFactory factory = Json.createBuilderFactory(config); final String applicationId = context.getProperty(APPLICATION_ID).evaluateAttributeExpressions().getValue(); final String hostname = context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue(); final ProcessGroupStatus status = context.getEventAccess().getControllerStatus(); if(status != null) { final Map<String,String> statusMetrics = metricsService.getMetrics(status, false); final Map<String,String> jvmMetrics = metricsService.getMetrics(virtualMachineMetrics); final MetricsBuilder metricsBuilder = new MetricsBuilder(factory); final OperatingSystemMXBean os = ManagementFactory.getOperatingSystemMXBean(); final double systemLoad = os.getSystemLoadAverage(); byte[] data; final Map<String, String> attributes = new HashMap<>(); if(context.getProperty(FORMAT).getValue().equals(AMBARI_FORMAT.getValue())) { final JsonObject metricsObject = metricsBuilder .applicationId(applicationId) .instanceId(status.getId()) .hostname(hostname) .timestamp(System.currentTimeMillis()) .addAllMetrics(statusMetrics) .addAllMetrics(jvmMetrics) .metric(MetricNames.CORES, String.valueOf(os.getAvailableProcessors())) .metric(MetricNames.LOAD1MN, String.valueOf(systemLoad >= 0 ? systemLoad : -1)) .build(); data = metricsObject.toString().getBytes(StandardCharsets.UTF_8); attributes.put(CoreAttributes.MIME_TYPE.key(), "application/json"); } else { final JsonObject metricsObject = metricsService.getMetrics(factory, status, virtualMachineMetrics, applicationId, status.getId(), hostname, System.currentTimeMillis(), os.getAvailableProcessors(), systemLoad >= 0 ? systemLoad : -1); data = getData(context, new ByteArrayInputStream(metricsObject.toString().getBytes(StandardCharsets.UTF_8)), attributes); } try { long start = System.nanoTime(); final Transaction transaction = getClient().createTransaction(TransferDirection.SEND); if (transaction == null) { getLogger().debug("All destination nodes are penalized; will attempt to send data later"); return; } final String transactionId = UUID.randomUUID().toString(); attributes.put("reporting.task.transaction.id", transactionId); attributes.put("reporting.task.name", getName()); attributes.put("reporting.task.uuid", getIdentifier()); attributes.put("reporting.task.type", this.getClass().getSimpleName()); transaction.send(data, attributes); transaction.confirm(); transaction.complete(); final long transferMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); getLogger().info("Successfully sent metrics to destination in {}ms; Transaction ID = {}", new Object[]{transferMillis, transactionId}); } catch (final Exception e) { throw new ProcessException("Failed to send metrics to destination due to:" + e.getMessage(), e); } } else { getLogger().error("No process group status to retrieve metrics"); } } SiteToSiteMetricsReportingTask(); @Override void onTrigger(final ReportingContext context);  }