@Test public void testEmptyRequest() throws IOException { String requestBody = ""; 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 error message", "Invalid request body", errorResponse.getErrors().get(0).getMessage()); assertEquals("Invalid tenant", TENANT, errorResponse.getErrors().get(0).getTenantId()); assertEquals("Invalid status", HttpResponseStatus.BAD_REQUEST, argument.getValue().getStatus()); }
@Override public void handle(ChannelHandlerContext ctx, FullHttpRequest request) { Tracker.getInstance().track(request); requestCount.inc(); final Timer.Context timerContext = handlerTimer.time(); long ingestTime = clock.now().getMillis(); String body = null; try { String submitterTenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER); body = request.content().toString(Constants.DEFAULT_CHARSET); List<AggregatedPayload> bundleList = createBundleList(body); if (bundleList.size() > 0) { MetricsCollection collection = new MetricsCollection(); List<ErrorResponse.ErrorData> errors = new ArrayList<ErrorResponse.ErrorData>(); int delayedMetricsCount = 0; int metricsCount = 0; for (AggregatedPayload bundle : bundleList) { List<ErrorResponse.ErrorData> bundleValidationErrors = bundle.getValidationErrors(); if (bundleValidationErrors.isEmpty()) { collection.add(PreaggregateConversions.buildMetricsCollection(bundle)); } else { errors.addAll(bundleValidationErrors); } if (bundle.hasDelayedMetrics(ingestTime)) { Tracker.getInstance().trackDelayedAggregatedMetricsTenant(bundle.getTenantId(), bundle.getTimestamp(), bundle.getDelayTime(ingestTime), bundle.getAllMetricNames()); bundle.markDelayMetricsReceived(ingestTime); delayedMetricsCount += bundle.getAllMetricNames().size(); } else { metricsCount += bundle.getAllMetricNames().size(); } } if (!errors.isEmpty() && collection.size() == 0) { DefaultHandler.sendErrorResponse(ctx, request, errors, HttpResponseStatus.BAD_REQUEST); return; } ListenableFuture<List<Boolean>> futures = processor.apply(collection); List<Boolean> persisteds = futures.get(timeout.getValue(), timeout.getUnit()); for (Boolean persisted : persisteds) { if (!persisted) { DefaultHandler.sendErrorResponse(ctx, request, "Internal error persisting data", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } } recordPerTenantMetrics(submitterTenantId, metricsCount, delayedMetricsCount); if (errors.isEmpty()) { DefaultHandler.sendResponse(ctx, request, null, HttpResponseStatus.OK); return; } else { DefaultHandler.sendErrorResponse(ctx, request, errors, HttpResponseStatus.MULTI_STATUS); return; } } else { DefaultHandler.sendResponse(ctx, request, "No valid metrics", HttpResponseStatus.BAD_REQUEST); return; } } catch (JsonParseException ex) { log.debug(String.format("BAD JSON: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (InvalidDataException ex) { log.debug(String.format("Invalid request body: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (TimeoutException ex) { DefaultHandler.sendErrorResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED); } catch (Exception ex) { log.debug(String.format("Exception processing: %s", body)); log.error("Other exception while trying to parse content", ex); DefaultHandler.sendErrorResponse(ctx, request, "Internal error saving data", HttpResponseStatus.INTERNAL_SERVER_ERROR); } finally { timerContext.stop(); requestCount.dec(); } }
HttpAggregatedMultiIngestionHandler implements HttpRequestHandler { @Override public void handle(ChannelHandlerContext ctx, FullHttpRequest request) { Tracker.getInstance().track(request); requestCount.inc(); final Timer.Context timerContext = handlerTimer.time(); long ingestTime = clock.now().getMillis(); String body = null; try { String submitterTenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER); body = request.content().toString(Constants.DEFAULT_CHARSET); List<AggregatedPayload> bundleList = createBundleList(body); if (bundleList.size() > 0) { MetricsCollection collection = new MetricsCollection(); List<ErrorResponse.ErrorData> errors = new ArrayList<ErrorResponse.ErrorData>(); int delayedMetricsCount = 0; int metricsCount = 0; for (AggregatedPayload bundle : bundleList) { List<ErrorResponse.ErrorData> bundleValidationErrors = bundle.getValidationErrors(); if (bundleValidationErrors.isEmpty()) { collection.add(PreaggregateConversions.buildMetricsCollection(bundle)); } else { errors.addAll(bundleValidationErrors); } if (bundle.hasDelayedMetrics(ingestTime)) { Tracker.getInstance().trackDelayedAggregatedMetricsTenant(bundle.getTenantId(), bundle.getTimestamp(), bundle.getDelayTime(ingestTime), bundle.getAllMetricNames()); bundle.markDelayMetricsReceived(ingestTime); delayedMetricsCount += bundle.getAllMetricNames().size(); } else { metricsCount += bundle.getAllMetricNames().size(); } } if (!errors.isEmpty() && collection.size() == 0) { DefaultHandler.sendErrorResponse(ctx, request, errors, HttpResponseStatus.BAD_REQUEST); return; } ListenableFuture<List<Boolean>> futures = processor.apply(collection); List<Boolean> persisteds = futures.get(timeout.getValue(), timeout.getUnit()); for (Boolean persisted : persisteds) { if (!persisted) { DefaultHandler.sendErrorResponse(ctx, request, "Internal error persisting data", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } } recordPerTenantMetrics(submitterTenantId, metricsCount, delayedMetricsCount); if (errors.isEmpty()) { DefaultHandler.sendResponse(ctx, request, null, HttpResponseStatus.OK); return; } else { DefaultHandler.sendErrorResponse(ctx, request, errors, HttpResponseStatus.MULTI_STATUS); return; } } else { DefaultHandler.sendResponse(ctx, request, "No valid metrics", HttpResponseStatus.BAD_REQUEST); return; } } catch (JsonParseException ex) { log.debug(String.format("BAD JSON: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (InvalidDataException ex) { log.debug(String.format("Invalid request body: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (TimeoutException ex) { DefaultHandler.sendErrorResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED); } catch (Exception ex) { log.debug(String.format("Exception processing: %s", body)); log.error("Other exception while trying to parse content", ex); DefaultHandler.sendErrorResponse(ctx, request, "Internal error saving data", HttpResponseStatus.INTERNAL_SERVER_ERROR); } finally { timerContext.stop(); requestCount.dec(); } } }
HttpAggregatedMultiIngestionHandler implements HttpRequestHandler { @Override public void handle(ChannelHandlerContext ctx, FullHttpRequest request) { Tracker.getInstance().track(request); requestCount.inc(); final Timer.Context timerContext = handlerTimer.time(); long ingestTime = clock.now().getMillis(); String body = null; try { String submitterTenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER); body = request.content().toString(Constants.DEFAULT_CHARSET); List<AggregatedPayload> bundleList = createBundleList(body); if (bundleList.size() > 0) { MetricsCollection collection = new MetricsCollection(); List<ErrorResponse.ErrorData> errors = new ArrayList<ErrorResponse.ErrorData>(); int delayedMetricsCount = 0; int metricsCount = 0; for (AggregatedPayload bundle : bundleList) { List<ErrorResponse.ErrorData> bundleValidationErrors = bundle.getValidationErrors(); if (bundleValidationErrors.isEmpty()) { collection.add(PreaggregateConversions.buildMetricsCollection(bundle)); } else { errors.addAll(bundleValidationErrors); } if (bundle.hasDelayedMetrics(ingestTime)) { Tracker.getInstance().trackDelayedAggregatedMetricsTenant(bundle.getTenantId(), bundle.getTimestamp(), bundle.getDelayTime(ingestTime), bundle.getAllMetricNames()); bundle.markDelayMetricsReceived(ingestTime); delayedMetricsCount += bundle.getAllMetricNames().size(); } else { metricsCount += bundle.getAllMetricNames().size(); } } if (!errors.isEmpty() && collection.size() == 0) { DefaultHandler.sendErrorResponse(ctx, request, errors, HttpResponseStatus.BAD_REQUEST); return; } ListenableFuture<List<Boolean>> futures = processor.apply(collection); List<Boolean> persisteds = futures.get(timeout.getValue(), timeout.getUnit()); for (Boolean persisted : persisteds) { if (!persisted) { DefaultHandler.sendErrorResponse(ctx, request, "Internal error persisting data", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } } recordPerTenantMetrics(submitterTenantId, metricsCount, delayedMetricsCount); if (errors.isEmpty()) { DefaultHandler.sendResponse(ctx, request, null, HttpResponseStatus.OK); return; } else { DefaultHandler.sendErrorResponse(ctx, request, errors, HttpResponseStatus.MULTI_STATUS); return; } } else { DefaultHandler.sendResponse(ctx, request, "No valid metrics", HttpResponseStatus.BAD_REQUEST); return; } } catch (JsonParseException ex) { log.debug(String.format("BAD JSON: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (InvalidDataException ex) { log.debug(String.format("Invalid request body: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (TimeoutException ex) { DefaultHandler.sendErrorResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED); } catch (Exception ex) { log.debug(String.format("Exception processing: %s", body)); log.error("Other exception while trying to parse content", ex); DefaultHandler.sendErrorResponse(ctx, request, "Internal error saving data", HttpResponseStatus.INTERNAL_SERVER_ERROR); } finally { timerContext.stop(); requestCount.dec(); } } HttpAggregatedMultiIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout);  HttpAggregatedMultiIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout, boolean enablePerTenantMetrics); }
HttpAggregatedMultiIngestionHandler implements HttpRequestHandler { @Override public void handle(ChannelHandlerContext ctx, FullHttpRequest request) { Tracker.getInstance().track(request); requestCount.inc(); final Timer.Context timerContext = handlerTimer.time(); long ingestTime = clock.now().getMillis(); String body = null; try { String submitterTenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER); body = request.content().toString(Constants.DEFAULT_CHARSET); List<AggregatedPayload> bundleList = createBundleList(body); if (bundleList.size() > 0) { MetricsCollection collection = new MetricsCollection(); List<ErrorResponse.ErrorData> errors = new ArrayList<ErrorResponse.ErrorData>(); int delayedMetricsCount = 0; int metricsCount = 0; for (AggregatedPayload bundle : bundleList) { List<ErrorResponse.ErrorData> bundleValidationErrors = bundle.getValidationErrors(); if (bundleValidationErrors.isEmpty()) { collection.add(PreaggregateConversions.buildMetricsCollection(bundle)); } else { errors.addAll(bundleValidationErrors); } if (bundle.hasDelayedMetrics(ingestTime)) { Tracker.getInstance().trackDelayedAggregatedMetricsTenant(bundle.getTenantId(), bundle.getTimestamp(), bundle.getDelayTime(ingestTime), bundle.getAllMetricNames()); bundle.markDelayMetricsReceived(ingestTime); delayedMetricsCount += bundle.getAllMetricNames().size(); } else { metricsCount += bundle.getAllMetricNames().size(); } } if (!errors.isEmpty() && collection.size() == 0) { DefaultHandler.sendErrorResponse(ctx, request, errors, HttpResponseStatus.BAD_REQUEST); return; } ListenableFuture<List<Boolean>> futures = processor.apply(collection); List<Boolean> persisteds = futures.get(timeout.getValue(), timeout.getUnit()); for (Boolean persisted : persisteds) { if (!persisted) { DefaultHandler.sendErrorResponse(ctx, request, "Internal error persisting data", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } } recordPerTenantMetrics(submitterTenantId, metricsCount, delayedMetricsCount); if (errors.isEmpty()) { DefaultHandler.sendResponse(ctx, request, null, HttpResponseStatus.OK); return; } else { DefaultHandler.sendErrorResponse(ctx, request, errors, HttpResponseStatus.MULTI_STATUS); return; } } else { DefaultHandler.sendResponse(ctx, request, "No valid metrics", HttpResponseStatus.BAD_REQUEST); return; } } catch (JsonParseException ex) { log.debug(String.format("BAD JSON: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (InvalidDataException ex) { log.debug(String.format("Invalid request body: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (TimeoutException ex) { DefaultHandler.sendErrorResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED); } catch (Exception ex) { log.debug(String.format("Exception processing: %s", body)); log.error("Other exception while trying to parse content", ex); DefaultHandler.sendErrorResponse(ctx, request, "Internal error saving data", HttpResponseStatus.INTERNAL_SERVER_ERROR); } finally { timerContext.stop(); requestCount.dec(); } } HttpAggregatedMultiIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout);  HttpAggregatedMultiIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout, boolean enablePerTenantMetrics); @Override void handle(ChannelHandlerContext ctx, FullHttpRequest request); static List<AggregatedPayload> createBundleList(String json); }
HttpAggregatedMultiIngestionHandler implements HttpRequestHandler { @Override public void handle(ChannelHandlerContext ctx, FullHttpRequest request) { Tracker.getInstance().track(request); requestCount.inc(); final Timer.Context timerContext = handlerTimer.time(); long ingestTime = clock.now().getMillis(); String body = null; try { String submitterTenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER); body = request.content().toString(Constants.DEFAULT_CHARSET); List<AggregatedPayload> bundleList = createBundleList(body); if (bundleList.size() > 0) { MetricsCollection collection = new MetricsCollection(); List<ErrorResponse.ErrorData> errors = new ArrayList<ErrorResponse.ErrorData>(); int delayedMetricsCount = 0; int metricsCount = 0; for (AggregatedPayload bundle : bundleList) { List<ErrorResponse.ErrorData> bundleValidationErrors = bundle.getValidationErrors(); if (bundleValidationErrors.isEmpty()) { collection.add(PreaggregateConversions.buildMetricsCollection(bundle)); } else { errors.addAll(bundleValidationErrors); } if (bundle.hasDelayedMetrics(ingestTime)) { Tracker.getInstance().trackDelayedAggregatedMetricsTenant(bundle.getTenantId(), bundle.getTimestamp(), bundle.getDelayTime(ingestTime), bundle.getAllMetricNames()); bundle.markDelayMetricsReceived(ingestTime); delayedMetricsCount += bundle.getAllMetricNames().size(); } else { metricsCount += bundle.getAllMetricNames().size(); } } if (!errors.isEmpty() && collection.size() == 0) { DefaultHandler.sendErrorResponse(ctx, request, errors, HttpResponseStatus.BAD_REQUEST); return; } ListenableFuture<List<Boolean>> futures = processor.apply(collection); List<Boolean> persisteds = futures.get(timeout.getValue(), timeout.getUnit()); for (Boolean persisted : persisteds) { if (!persisted) { DefaultHandler.sendErrorResponse(ctx, request, "Internal error persisting data", HttpResponseStatus.INTERNAL_SERVER_ERROR); return; } } recordPerTenantMetrics(submitterTenantId, metricsCount, delayedMetricsCount); if (errors.isEmpty()) { DefaultHandler.sendResponse(ctx, request, null, HttpResponseStatus.OK); return; } else { DefaultHandler.sendErrorResponse(ctx, request, errors, HttpResponseStatus.MULTI_STATUS); return; } } else { DefaultHandler.sendResponse(ctx, request, "No valid metrics", HttpResponseStatus.BAD_REQUEST); return; } } catch (JsonParseException ex) { log.debug(String.format("BAD JSON: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (InvalidDataException ex) { log.debug(String.format("Invalid request body: %s", body)); DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST); } catch (TimeoutException ex) { DefaultHandler.sendErrorResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED); } catch (Exception ex) { log.debug(String.format("Exception processing: %s", body)); log.error("Other exception while trying to parse content", ex); DefaultHandler.sendErrorResponse(ctx, request, "Internal error saving data", HttpResponseStatus.INTERNAL_SERVER_ERROR); } finally { timerContext.stop(); requestCount.dec(); } } HttpAggregatedMultiIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout);  HttpAggregatedMultiIngestionHandler(HttpMetricsIngestionServer.Processor processor, TimeValue timeout, boolean enablePerTenantMetrics); @Override void handle(ChannelHandlerContext ctx, FullHttpRequest request); static List<AggregatedPayload> createBundleList(String json);  }