@Test public void testSingleMetricInvalidMetricName() throws IOException { String metricName = ""; String singleMetric = createRequestBody(metricName, new DefaultClockImpl().now().getMillis(), 24 * 60 * 60, 1); String requestBody = "[" + singleMetric + "]"; FullHttpRequest request = createIngestRequest(requestBody); ArgumentCaptor<FullHttpResponse> argument = ArgumentCaptor.forClass(FullHttpResponse.class); handler.handle(context, request); verify(channel).write(argument.capture()); String errorResponseBody = argument.getValue().content().toString(Charset.defaultCharset()); ErrorResponse errorResponse = getErrorResponse(errorResponseBody); assertEquals("Number of errors invalid", 1, errorResponse.getErrors().size()); assertEquals("Invalid tenant", TENANT, errorResponse.getErrors().get(0).getTenantId()); assertEquals("Invalid metric name", metricName, errorResponse.getErrors().get(0).getMetricName()); assertEquals("Invalid status", HttpResponseStatus.BAD_REQUEST, argument.getValue().getStatus()); assertEquals("Invalid error source", "metricName", errorResponse.getErrors().get(0).getSource()); assertEquals("Invalid error message", "may not be empty", errorResponse.getErrors().get(0).getMessage()); }
@Override public void handle(ChannelHandlerContext ctx, FullHttpRequest request) { try { requestsReceived.mark(); Tracker.getInstance().track(request); requestCount.inc(); final String tenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER); JSONMetricsContainer jsonMetricsContainer; List<Metric> validMetrics; final Timer.Context jsonTimerContext = jsonTimer.time(); final String body = request.content().toString(Constants.DEFAULT_CHARSET); try { jsonMetricsContainer = createContainer(body, tenantId); if (jsonMetricsContainer.areDelayedMetricsPresent()) { Tracker.getInstance().trackDelayedMetricsTenant(tenantId, jsonMetricsContainer.getDelayedMetrics()); } validMetrics = jsonMetricsContainer.getValidMetrics(); forceTTLsIfConfigured(validMetrics); } catch (JsonParseException e) { log.warn("Exception parsing content", e); DefaultHandler.sendErrorResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST); return; } catch (JsonMappingException e) { log.warn("Exception parsing content", e); DefaultHandler.sendErrorResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST); return; } catch (InvalidDataException ex) { log.warn(ctx.channel().remoteAddress() + " " + ex.getMessage()); DefaultHandler.sendErrorResponse(ctx, request, "Invalid data " + ex.getMessage(), HttpResponseStatus.BAD_REQUEST); return; } catch (IOException e) { log.warn("IO Exception parsing content", e); DefaultHandler.sendErrorResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST); return; } catch (Exception e) { log.warn("Other exception while trying to parse content", e); DefaultHandler.sendErrorResponse(ctx, request, "Failed parsing content", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } finally { jsonTimerContext.stop(); } List<ErrorResponse.ErrorData> validationErrors = jsonMetricsContainer.getValidationErrors(); if (validMetrics.isEmpty()) { log.warn(ctx.channel().remoteAddress() + " No valid metrics"); if (validationErrors.isEmpty()) { DefaultHandler.sendErrorResponse(ctx, request, "No valid metrics", HttpResponseStatus.BAD_REQUEST); } else { DefaultHandler.sendErrorResponse(ctx, request, validationErrors, HttpResponseStatus.BAD_REQUEST); } return; } final MetricsCollection collection = new MetricsCollection(); collection.add(new ArrayList<IMetric>(validMetrics)); final Timer.Context persistingTimerContext = persistingTimer.time(); try { ListenableFuture<List<Boolean>> futures = processor.apply(collection); List<Boolean> persisteds = futures.get(timeout.getValue(), timeout.getUnit()); for (Boolean persisted : persisteds) { if (!persisted) { log.warn("Trouble persisting metrics:"); log.warn(String.format("%s", Arrays.toString(validMetrics.toArray()))); DefaultHandler.sendErrorResponse(ctx, request, "Persisted failed for metrics", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } } recordPerTenantMetrics(tenantId, jsonMetricsContainer.getNonDelayedMetricsCount(), jsonMetricsContainer.getDelayedMetricsCount()); if( !validationErrors.isEmpty() ) { DefaultHandler.sendErrorResponse(ctx, request, validationErrors, HttpResponseStatus.MULTI_STATUS); } else { DefaultHandler.sendResponse(ctx, request, null, HttpResponseStatus.OK); } } catch (TimeoutException e) { DefaultHandler.sendErrorResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED); } catch (Exception e) { log.error("Exception persisting metrics", e); DefaultHandler.sendErrorResponse(ctx, request, "Error persisting metrics", HttpResponseStatus.INTERNAL_SERVER_ERROR); } finally { persistingTimerContext.stop(); } } finally { requestCount.dec(); } }
HttpMetricsIngestionHandler implements HttpRequestHandler { @Override public void handle(ChannelHandlerContext ctx, FullHttpRequest request) { try { requestsReceived.mark(); Tracker.getInstance().track(request); requestCount.inc(); final String tenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER); JSONMetricsContainer jsonMetricsContainer; List<Metric> validMetrics; final Timer.Context jsonTimerContext = jsonTimer.time(); final String body = request.content().toString(Constants.DEFAULT_CHARSET); try { jsonMetricsContainer = createContainer(body, tenantId); if (jsonMetricsContainer.areDelayedMetricsPresent()) { Tracker.getInstance().trackDelayedMetricsTenant(tenantId, jsonMetricsContainer.getDelayedMetrics()); } validMetrics = jsonMetricsContainer.getValidMetrics(); forceTTLsIfConfigured(validMetrics); } catch (JsonParseException e) { log.warn("Exception parsing content", e); DefaultHandler.sendErrorResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST); return; } catch (JsonMappingException e) { log.warn("Exception parsing content", e); DefaultHandler.sendErrorResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST); return; } catch (InvalidDataException ex) { log.warn(ctx.channel().remoteAddress() + " " + ex.getMessage()); DefaultHandler.sendErrorResponse(ctx, request, "Invalid data " + ex.getMessage(), HttpResponseStatus.BAD_REQUEST); return; } catch (IOException e) { log.warn("IO Exception parsing content", e); DefaultHandler.sendErrorResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST); return; } catch (Exception e) { log.warn("Other exception while trying to parse content", e); DefaultHandler.sendErrorResponse(ctx, request, "Failed parsing content", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } finally { jsonTimerContext.stop(); } List<ErrorResponse.ErrorData> validationErrors = jsonMetricsContainer.getValidationErrors(); if (validMetrics.isEmpty()) { log.warn(ctx.channel().remoteAddress() + " No valid metrics"); if (validationErrors.isEmpty()) { DefaultHandler.sendErrorResponse(ctx, request, "No valid metrics", HttpResponseStatus.BAD_REQUEST); } else { DefaultHandler.sendErrorResponse(ctx, request, validationErrors, HttpResponseStatus.BAD_REQUEST); } return; } final MetricsCollection collection = new MetricsCollection(); collection.add(new ArrayList<IMetric>(validMetrics)); final Timer.Context persistingTimerContext = persistingTimer.time(); try { ListenableFuture<List<Boolean>> futures = processor.apply(collection); List<Boolean> persisteds = futures.get(timeout.getValue(), timeout.getUnit()); for (Boolean persisted : persisteds) { if (!persisted) { log.warn("Trouble persisting metrics:"); log.warn(String.format("%s", Arrays.toString(validMetrics.toArray()))); DefaultHandler.sendErrorResponse(ctx, request, "Persisted failed for metrics", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } } recordPerTenantMetrics(tenantId, jsonMetricsContainer.getNonDelayedMetricsCount(), jsonMetricsContainer.getDelayedMetricsCount()); if( !validationErrors.isEmpty() ) { DefaultHandler.sendErrorResponse(ctx, request, validationErrors, HttpResponseStatus.MULTI_STATUS); } else { DefaultHandler.sendResponse(ctx, request, null, HttpResponseStatus.OK); } } catch (TimeoutException e) { DefaultHandler.sendErrorResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED); } catch (Exception e) { log.error("Exception persisting metrics", e); DefaultHandler.sendErrorResponse(ctx, request, "Error persisting metrics", HttpResponseStatus.INTERNAL_SERVER_ERROR); } finally { persistingTimerContext.stop(); } } finally { requestCount.dec(); } } }
HttpMetricsIngestionHandler implements HttpRequestHandler { @Override public void handle(ChannelHandlerContext ctx, FullHttpRequest request) { try { requestsReceived.mark(); Tracker.getInstance().track(request); requestCount.inc(); final String tenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER); JSONMetricsContainer jsonMetricsContainer; List<Metric> validMetrics; final Timer.Context jsonTimerContext = jsonTimer.time(); final String body = request.content().toString(Constants.DEFAULT_CHARSET); try { jsonMetricsContainer = createContainer(body, tenantId); if (jsonMetricsContainer.areDelayedMetricsPresent()) { Tracker.getInstance().trackDelayedMetricsTenant(tenantId, jsonMetricsContainer.getDelayedMetrics()); } validMetrics = jsonMetricsContainer.getValidMetrics(); forceTTLsIfConfigured(validMetrics); } catch (JsonParseException e) { log.warn("Exception parsing content", e); DefaultHandler.sendErrorResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST); return; } catch (JsonMappingException e) { log.warn("Exception parsing content", e); DefaultHandler.sendErrorResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST); return; } catch (InvalidDataException ex) { log.warn(ctx.channel().remoteAddress() + " " + ex.getMessage()); DefaultHandler.sendErrorResponse(ctx, request, "Invalid data " + ex.getMessage(), HttpResponseStatus.BAD_REQUEST); return; } catch (IOException e) { log.warn("IO Exception parsing content", e); DefaultHandler.sendErrorResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST); return; } catch (Exception e) { log.warn("Other exception while trying to parse content", e); DefaultHandler.sendErrorResponse(ctx, request, "Failed parsing content", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } finally { jsonTimerContext.stop(); } List<ErrorResponse.ErrorData> validationErrors = jsonMetricsContainer.getValidationErrors(); if (validMetrics.isEmpty()) { log.warn(ctx.channel().remoteAddress() + " No valid metrics"); if (validationErrors.isEmpty()) { DefaultHandler.sendErrorResponse(ctx, request, "No valid metrics", HttpResponseStatus.BAD_REQUEST); } else { DefaultHandler.sendErrorResponse(ctx, request, validationErrors, HttpResponseStatus.BAD_REQUEST); } return; } final MetricsCollection collection = new MetricsCollection(); collection.add(new ArrayList<IMetric>(validMetrics)); final Timer.Context persistingTimerContext = persistingTimer.time(); try { ListenableFuture<List<Boolean>> futures = processor.apply(collection); List<Boolean> persisteds = futures.get(timeout.getValue(), timeout.getUnit()); for (Boolean persisted : persisteds) { if (!persisted) { log.warn("Trouble persisting metrics:"); log.warn(String.format("%s", Arrays.toString(validMetrics.toArray()))); DefaultHandler.sendErrorResponse(ctx, request, "Persisted failed for metrics", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } } recordPerTenantMetrics(tenantId, jsonMetricsContainer.getNonDelayedMetricsCount(), jsonMetricsContainer.getDelayedMetricsCount()); if( !validationErrors.isEmpty() ) { DefaultHandler.sendErrorResponse(ctx, request, validationErrors, HttpResponseStatus.MULTI_STATUS); } else { DefaultHandler.sendResponse(ctx, request, null, HttpResponseStatus.OK); } } catch (TimeoutException e) { DefaultHandler.sendErrorResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED); } catch (Exception e) { log.error("Exception persisting metrics", e); DefaultHandler.sendErrorResponse(ctx, request, "Error persisting metrics", HttpResponseStatus.INTERNAL_SERVER_ERROR); } finally { persistingTimerContext.stop(); } } finally { requestCount.dec(); } } HttpMetricsIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout);  HttpMetricsIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout, boolean enablePerTenantMetrics); }
HttpMetricsIngestionHandler implements HttpRequestHandler { @Override public void handle(ChannelHandlerContext ctx, FullHttpRequest request) { try { requestsReceived.mark(); Tracker.getInstance().track(request); requestCount.inc(); final String tenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER); JSONMetricsContainer jsonMetricsContainer; List<Metric> validMetrics; final Timer.Context jsonTimerContext = jsonTimer.time(); final String body = request.content().toString(Constants.DEFAULT_CHARSET); try { jsonMetricsContainer = createContainer(body, tenantId); if (jsonMetricsContainer.areDelayedMetricsPresent()) { Tracker.getInstance().trackDelayedMetricsTenant(tenantId, jsonMetricsContainer.getDelayedMetrics()); } validMetrics = jsonMetricsContainer.getValidMetrics(); forceTTLsIfConfigured(validMetrics); } catch (JsonParseException e) { log.warn("Exception parsing content", e); DefaultHandler.sendErrorResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST); return; } catch (JsonMappingException e) { log.warn("Exception parsing content", e); DefaultHandler.sendErrorResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST); return; } catch (InvalidDataException ex) { log.warn(ctx.channel().remoteAddress() + " " + ex.getMessage()); DefaultHandler.sendErrorResponse(ctx, request, "Invalid data " + ex.getMessage(), HttpResponseStatus.BAD_REQUEST); return; } catch (IOException e) { log.warn("IO Exception parsing content", e); DefaultHandler.sendErrorResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST); return; } catch (Exception e) { log.warn("Other exception while trying to parse content", e); DefaultHandler.sendErrorResponse(ctx, request, "Failed parsing content", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } finally { jsonTimerContext.stop(); } List<ErrorResponse.ErrorData> validationErrors = jsonMetricsContainer.getValidationErrors(); if (validMetrics.isEmpty()) { log.warn(ctx.channel().remoteAddress() + " No valid metrics"); if (validationErrors.isEmpty()) { DefaultHandler.sendErrorResponse(ctx, request, "No valid metrics", HttpResponseStatus.BAD_REQUEST); } else { DefaultHandler.sendErrorResponse(ctx, request, validationErrors, HttpResponseStatus.BAD_REQUEST); } return; } final MetricsCollection collection = new MetricsCollection(); collection.add(new ArrayList<IMetric>(validMetrics)); final Timer.Context persistingTimerContext = persistingTimer.time(); try { ListenableFuture<List<Boolean>> futures = processor.apply(collection); List<Boolean> persisteds = futures.get(timeout.getValue(), timeout.getUnit()); for (Boolean persisted : persisteds) { if (!persisted) { log.warn("Trouble persisting metrics:"); log.warn(String.format("%s", Arrays.toString(validMetrics.toArray()))); DefaultHandler.sendErrorResponse(ctx, request, "Persisted failed for metrics", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } } recordPerTenantMetrics(tenantId, jsonMetricsContainer.getNonDelayedMetricsCount(), jsonMetricsContainer.getDelayedMetricsCount()); if( !validationErrors.isEmpty() ) { DefaultHandler.sendErrorResponse(ctx, request, validationErrors, HttpResponseStatus.MULTI_STATUS); } else { DefaultHandler.sendResponse(ctx, request, null, HttpResponseStatus.OK); } } catch (TimeoutException e) { DefaultHandler.sendErrorResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED); } catch (Exception e) { log.error("Exception persisting metrics", e); DefaultHandler.sendErrorResponse(ctx, request, "Error persisting metrics", HttpResponseStatus.INTERNAL_SERVER_ERROR); } finally { persistingTimerContext.stop(); } } finally { requestCount.dec(); } } HttpMetricsIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout);  HttpMetricsIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout, boolean enablePerTenantMetrics); static String getResponseBody( List<String> errors ); @Override void handle(ChannelHandlerContext ctx, FullHttpRequest request); }
HttpMetricsIngestionHandler implements HttpRequestHandler { @Override public void handle(ChannelHandlerContext ctx, FullHttpRequest request) { try { requestsReceived.mark(); Tracker.getInstance().track(request); requestCount.inc(); final String tenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER); JSONMetricsContainer jsonMetricsContainer; List<Metric> validMetrics; final Timer.Context jsonTimerContext = jsonTimer.time(); final String body = request.content().toString(Constants.DEFAULT_CHARSET); try { jsonMetricsContainer = createContainer(body, tenantId); if (jsonMetricsContainer.areDelayedMetricsPresent()) { Tracker.getInstance().trackDelayedMetricsTenant(tenantId, jsonMetricsContainer.getDelayedMetrics()); } validMetrics = jsonMetricsContainer.getValidMetrics(); forceTTLsIfConfigured(validMetrics); } catch (JsonParseException e) { log.warn("Exception parsing content", e); DefaultHandler.sendErrorResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST); return; } catch (JsonMappingException e) { log.warn("Exception parsing content", e); DefaultHandler.sendErrorResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST); return; } catch (InvalidDataException ex) { log.warn(ctx.channel().remoteAddress() + " " + ex.getMessage()); DefaultHandler.sendErrorResponse(ctx, request, "Invalid data " + ex.getMessage(), HttpResponseStatus.BAD_REQUEST); return; } catch (IOException e) { log.warn("IO Exception parsing content", e); DefaultHandler.sendErrorResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST); return; } catch (Exception e) { log.warn("Other exception while trying to parse content", e); DefaultHandler.sendErrorResponse(ctx, request, "Failed parsing content", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } finally { jsonTimerContext.stop(); } List<ErrorResponse.ErrorData> validationErrors = jsonMetricsContainer.getValidationErrors(); if (validMetrics.isEmpty()) { log.warn(ctx.channel().remoteAddress() + " No valid metrics"); if (validationErrors.isEmpty()) { DefaultHandler.sendErrorResponse(ctx, request, "No valid metrics", HttpResponseStatus.BAD_REQUEST); } else { DefaultHandler.sendErrorResponse(ctx, request, validationErrors, HttpResponseStatus.BAD_REQUEST); } return; } final MetricsCollection collection = new MetricsCollection(); collection.add(new ArrayList<IMetric>(validMetrics)); final Timer.Context persistingTimerContext = persistingTimer.time(); try { ListenableFuture<List<Boolean>> futures = processor.apply(collection); List<Boolean> persisteds = futures.get(timeout.getValue(), timeout.getUnit()); for (Boolean persisted : persisteds) { if (!persisted) { log.warn("Trouble persisting metrics:"); log.warn(String.format("%s", Arrays.toString(validMetrics.toArray()))); DefaultHandler.sendErrorResponse(ctx, request, "Persisted failed for metrics", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } } recordPerTenantMetrics(tenantId, jsonMetricsContainer.getNonDelayedMetricsCount(), jsonMetricsContainer.getDelayedMetricsCount()); if( !validationErrors.isEmpty() ) { DefaultHandler.sendErrorResponse(ctx, request, validationErrors, HttpResponseStatus.MULTI_STATUS); } else { DefaultHandler.sendResponse(ctx, request, null, HttpResponseStatus.OK); } } catch (TimeoutException e) { DefaultHandler.sendErrorResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED); } catch (Exception e) { log.error("Exception persisting metrics", e); DefaultHandler.sendErrorResponse(ctx, request, "Error persisting metrics", HttpResponseStatus.INTERNAL_SERVER_ERROR); } finally { persistingTimerContext.stop(); } } finally { requestCount.dec(); } } HttpMetricsIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout);  HttpMetricsIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout, boolean enablePerTenantMetrics); static String getResponseBody( List<String> errors ); @Override void handle(ChannelHandlerContext ctx, FullHttpRequest request); static final String ERROR_HEADER; }