@Test public void testOnTrigger() throws InitializationException, IOException { final String metricsUrl = "http: final String applicationId = "NIFI"; final String hostName = "localhost"; final Client client = Mockito.mock(Client.class); final WebTarget target = Mockito.mock(WebTarget.class); final Invocation.Builder builder = Mockito.mock(Invocation.Builder.class); final Response response = Mockito.mock(Response.class); Mockito.when(response.getStatus()).thenReturn(200); Mockito.when(client.target(metricsUrl)).thenReturn(target); Mockito.when(target.request()).thenReturn(builder); Mockito.when(builder.post(Matchers.any(Entity.class))).thenReturn(response); final ComponentLog logger = Mockito.mock(ComponentLog.class); final ReportingInitializationContext initContext = Mockito.mock(ReportingInitializationContext.class); Mockito.when(initContext.getIdentifier()).thenReturn(UUID.randomUUID().toString()); Mockito.when(initContext.getLogger()).thenReturn(logger); final ConfigurationContext configurationContext = Mockito.mock(ConfigurationContext.class); final ReportingContext context = Mockito.mock(ReportingContext.class); Mockito.when(context.getProperty(AmbariReportingTask.METRICS_COLLECTOR_URL)) .thenReturn(new MockPropertyValue(metricsUrl)); Mockito.when(context.getProperty(AmbariReportingTask.APPLICATION_ID)) .thenReturn(new MockPropertyValue(applicationId)); Mockito.when(context.getProperty(AmbariReportingTask.HOSTNAME)) .thenReturn(new MockPropertyValue(hostName)); Mockito.when(context.getProperty(AmbariReportingTask.PROCESS_GROUP_ID)) .thenReturn(new MockPropertyValue("1234")); final EventAccess eventAccess = Mockito.mock(EventAccess.class); Mockito.when(context.getEventAccess()).thenReturn(eventAccess); Mockito.when(eventAccess.getControllerStatus()).thenReturn(status); final AmbariReportingTask task = new TestableAmbariReportingTask(client); task.initialize(initContext); task.setup(configurationContext); task.onTrigger(context); }
@Override public void onTrigger(final ReportingContext context) { final String metricsCollectorUrl = context.getProperty(METRICS_COLLECTOR_URL).evaluateAttributeExpressions().getValue(); final String applicationId = context.getProperty(APPLICATION_ID).evaluateAttributeExpressions().getValue(); final String hostname = context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue(); final boolean pgIdIsSet = context.getProperty(PROCESS_GROUP_ID).isSet(); final String processGroupId = pgIdIsSet ? context.getProperty(PROCESS_GROUP_ID).evaluateAttributeExpressions().getValue() : null; final long start = System.currentTimeMillis(); if (previousMetrics != null) { final WebTarget metricsTarget = client.target(metricsCollectorUrl); final Invocation.Builder invocation = metricsTarget.request(); final Entity<String> entity = Entity.json(previousMetrics.toString()); getLogger().debug("Sending metrics {} to Ambari", new Object[]{entity.getEntity()}); final Response response = invocation.post(entity); if (response.getStatus() == Response.Status.OK.getStatusCode()) { final long completedMillis = TimeUnit.NANOSECONDS.toMillis(System.currentTimeMillis() - start); getLogger().info("Successfully sent metrics to Ambari in {} ms", new Object[]{completedMillis}); } else { final String responseEntity = response.hasEntity() ? response.readEntity(String.class) : "unknown error"; getLogger().error("Error sending metrics to Ambari due to {} - {}", new Object[]{response.getStatus(), responseEntity}); } } final ProcessGroupStatus status = processGroupId == null ? context.getEventAccess().getControllerStatus() : context.getEventAccess().getGroupStatus(processGroupId); if(status != null) { final Map<String,String> statusMetrics = metricsService.getMetrics(status, pgIdIsSet); final Map<String,String> jvmMetrics = metricsService.getMetrics(virtualMachineMetrics); final MetricsBuilder metricsBuilder = new MetricsBuilder(factory); final JsonObject metricsObject = metricsBuilder .applicationId(applicationId) .instanceId(status.getId()) .hostname(hostname) .timestamp(start) .addAllMetrics(statusMetrics) .addAllMetrics(jvmMetrics) .build(); previousMetrics = metricsObject; } else { getLogger().error("No process group status with ID = {}", new Object[]{processGroupId}); previousMetrics = null; } }
AmbariReportingTask extends AbstractReportingTask { @Override public void onTrigger(final ReportingContext context) { final String metricsCollectorUrl = context.getProperty(METRICS_COLLECTOR_URL).evaluateAttributeExpressions().getValue(); final String applicationId = context.getProperty(APPLICATION_ID).evaluateAttributeExpressions().getValue(); final String hostname = context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue(); final boolean pgIdIsSet = context.getProperty(PROCESS_GROUP_ID).isSet(); final String processGroupId = pgIdIsSet ? context.getProperty(PROCESS_GROUP_ID).evaluateAttributeExpressions().getValue() : null; final long start = System.currentTimeMillis(); if (previousMetrics != null) { final WebTarget metricsTarget = client.target(metricsCollectorUrl); final Invocation.Builder invocation = metricsTarget.request(); final Entity<String> entity = Entity.json(previousMetrics.toString()); getLogger().debug("Sending metrics {} to Ambari", new Object[]{entity.getEntity()}); final Response response = invocation.post(entity); if (response.getStatus() == Response.Status.OK.getStatusCode()) { final long completedMillis = TimeUnit.NANOSECONDS.toMillis(System.currentTimeMillis() - start); getLogger().info("Successfully sent metrics to Ambari in {} ms", new Object[]{completedMillis}); } else { final String responseEntity = response.hasEntity() ? response.readEntity(String.class) : "unknown error"; getLogger().error("Error sending metrics to Ambari due to {} - {}", new Object[]{response.getStatus(), responseEntity}); } } final ProcessGroupStatus status = processGroupId == null ? context.getEventAccess().getControllerStatus() : context.getEventAccess().getGroupStatus(processGroupId); if(status != null) { final Map<String,String> statusMetrics = metricsService.getMetrics(status, pgIdIsSet); final Map<String,String> jvmMetrics = metricsService.getMetrics(virtualMachineMetrics); final MetricsBuilder metricsBuilder = new MetricsBuilder(factory); final JsonObject metricsObject = metricsBuilder .applicationId(applicationId) .instanceId(status.getId()) .hostname(hostname) .timestamp(start) .addAllMetrics(statusMetrics) .addAllMetrics(jvmMetrics) .build(); previousMetrics = metricsObject; } else { getLogger().error("No process group status with ID = {}", new Object[]{processGroupId}); previousMetrics = null; } } }
AmbariReportingTask extends AbstractReportingTask { @Override public void onTrigger(final ReportingContext context) { final String metricsCollectorUrl = context.getProperty(METRICS_COLLECTOR_URL).evaluateAttributeExpressions().getValue(); final String applicationId = context.getProperty(APPLICATION_ID).evaluateAttributeExpressions().getValue(); final String hostname = context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue(); final boolean pgIdIsSet = context.getProperty(PROCESS_GROUP_ID).isSet(); final String processGroupId = pgIdIsSet ? context.getProperty(PROCESS_GROUP_ID).evaluateAttributeExpressions().getValue() : null; final long start = System.currentTimeMillis(); if (previousMetrics != null) { final WebTarget metricsTarget = client.target(metricsCollectorUrl); final Invocation.Builder invocation = metricsTarget.request(); final Entity<String> entity = Entity.json(previousMetrics.toString()); getLogger().debug("Sending metrics {} to Ambari", new Object[]{entity.getEntity()}); final Response response = invocation.post(entity); if (response.getStatus() == Response.Status.OK.getStatusCode()) { final long completedMillis = TimeUnit.NANOSECONDS.toMillis(System.currentTimeMillis() - start); getLogger().info("Successfully sent metrics to Ambari in {} ms", new Object[]{completedMillis}); } else { final String responseEntity = response.hasEntity() ? response.readEntity(String.class) : "unknown error"; getLogger().error("Error sending metrics to Ambari due to {} - {}", new Object[]{response.getStatus(), responseEntity}); } } final ProcessGroupStatus status = processGroupId == null ? context.getEventAccess().getControllerStatus() : context.getEventAccess().getGroupStatus(processGroupId); if(status != null) { final Map<String,String> statusMetrics = metricsService.getMetrics(status, pgIdIsSet); final Map<String,String> jvmMetrics = metricsService.getMetrics(virtualMachineMetrics); final MetricsBuilder metricsBuilder = new MetricsBuilder(factory); final JsonObject metricsObject = metricsBuilder .applicationId(applicationId) .instanceId(status.getId()) .hostname(hostname) .timestamp(start) .addAllMetrics(statusMetrics) .addAllMetrics(jvmMetrics) .build(); previousMetrics = metricsObject; } else { getLogger().error("No process group status with ID = {}", new Object[]{processGroupId}); previousMetrics = null; } }  }
AmbariReportingTask extends AbstractReportingTask { @Override public void onTrigger(final ReportingContext context) { final String metricsCollectorUrl = context.getProperty(METRICS_COLLECTOR_URL).evaluateAttributeExpressions().getValue(); final String applicationId = context.getProperty(APPLICATION_ID).evaluateAttributeExpressions().getValue(); final String hostname = context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue(); final boolean pgIdIsSet = context.getProperty(PROCESS_GROUP_ID).isSet(); final String processGroupId = pgIdIsSet ? context.getProperty(PROCESS_GROUP_ID).evaluateAttributeExpressions().getValue() : null; final long start = System.currentTimeMillis(); if (previousMetrics != null) { final WebTarget metricsTarget = client.target(metricsCollectorUrl); final Invocation.Builder invocation = metricsTarget.request(); final Entity<String> entity = Entity.json(previousMetrics.toString()); getLogger().debug("Sending metrics {} to Ambari", new Object[]{entity.getEntity()}); final Response response = invocation.post(entity); if (response.getStatus() == Response.Status.OK.getStatusCode()) { final long completedMillis = TimeUnit.NANOSECONDS.toMillis(System.currentTimeMillis() - start); getLogger().info("Successfully sent metrics to Ambari in {} ms", new Object[]{completedMillis}); } else { final String responseEntity = response.hasEntity() ? response.readEntity(String.class) : "unknown error"; getLogger().error("Error sending metrics to Ambari due to {} - {}", new Object[]{response.getStatus(), responseEntity}); } } final ProcessGroupStatus status = processGroupId == null ? context.getEventAccess().getControllerStatus() : context.getEventAccess().getGroupStatus(processGroupId); if(status != null) { final Map<String,String> statusMetrics = metricsService.getMetrics(status, pgIdIsSet); final Map<String,String> jvmMetrics = metricsService.getMetrics(virtualMachineMetrics); final MetricsBuilder metricsBuilder = new MetricsBuilder(factory); final JsonObject metricsObject = metricsBuilder .applicationId(applicationId) .instanceId(status.getId()) .hostname(hostname) .timestamp(start) .addAllMetrics(statusMetrics) .addAllMetrics(jvmMetrics) .build(); previousMetrics = metricsObject; } else { getLogger().error("No process group status with ID = {}", new Object[]{processGroupId}); previousMetrics = null; } }  @OnScheduled void setup(final ConfigurationContext context); @Override void onTrigger(final ReportingContext context); }
AmbariReportingTask extends AbstractReportingTask { @Override public void onTrigger(final ReportingContext context) { final String metricsCollectorUrl = context.getProperty(METRICS_COLLECTOR_URL).evaluateAttributeExpressions().getValue(); final String applicationId = context.getProperty(APPLICATION_ID).evaluateAttributeExpressions().getValue(); final String hostname = context.getProperty(HOSTNAME).evaluateAttributeExpressions().getValue(); final boolean pgIdIsSet = context.getProperty(PROCESS_GROUP_ID).isSet(); final String processGroupId = pgIdIsSet ? context.getProperty(PROCESS_GROUP_ID).evaluateAttributeExpressions().getValue() : null; final long start = System.currentTimeMillis(); if (previousMetrics != null) { final WebTarget metricsTarget = client.target(metricsCollectorUrl); final Invocation.Builder invocation = metricsTarget.request(); final Entity<String> entity = Entity.json(previousMetrics.toString()); getLogger().debug("Sending metrics {} to Ambari", new Object[]{entity.getEntity()}); final Response response = invocation.post(entity); if (response.getStatus() == Response.Status.OK.getStatusCode()) { final long completedMillis = TimeUnit.NANOSECONDS.toMillis(System.currentTimeMillis() - start); getLogger().info("Successfully sent metrics to Ambari in {} ms", new Object[]{completedMillis}); } else { final String responseEntity = response.hasEntity() ? response.readEntity(String.class) : "unknown error"; getLogger().error("Error sending metrics to Ambari due to {} - {}", new Object[]{response.getStatus(), responseEntity}); } } final ProcessGroupStatus status = processGroupId == null ? context.getEventAccess().getControllerStatus() : context.getEventAccess().getGroupStatus(processGroupId); if(status != null) { final Map<String,String> statusMetrics = metricsService.getMetrics(status, pgIdIsSet); final Map<String,String> jvmMetrics = metricsService.getMetrics(virtualMachineMetrics); final MetricsBuilder metricsBuilder = new MetricsBuilder(factory); final JsonObject metricsObject = metricsBuilder .applicationId(applicationId) .instanceId(status.getId()) .hostname(hostname) .timestamp(start) .addAllMetrics(statusMetrics) .addAllMetrics(jvmMetrics) .build(); previousMetrics = metricsObject; } else { getLogger().error("No process group status with ID = {}", new Object[]{processGroupId}); previousMetrics = null; } }  @OnScheduled void setup(final ConfigurationContext context); @Override void onTrigger(final ReportingContext context);  }