ElasticSearchAnnotationService extends DefaultService implements AnnotationStorageService { @Override public void putAnnotations(List<Annotation> annotations) { String requestUrl = "_bulk"; try { AnnotationRecordList indexAnnotationRecordList = new AnnotationRecordList(annotations, idgenHashAlgo); String requestBody = annotationMapper.writeValueAsString(indexAnnotationRecordList); Set<Annotation> failedRecords = new HashSet<>(); ElasticSearchUtils.PutResponse putResponse = ElasticSearchUtils.performESRequest(esRestClient, requestUrl, requestBody); if(putResponse.isErrors()) { for(ElasticSearchUtils.PutResponse.Item item : putResponse.getItems()) { if (item.getIndex() != null && item.getIndex().getStatus() != HttpStatus.SC_CREATED) { logger.warn("Failed to add record {} to index. Reason: {}", indexAnnotationRecordList.getRecord(item.getIndex().get_id()), annotationMapper.writeValueAsString(item.getIndex().getError())); failedRecords.add(indexAnnotationRecordList.getRecord(item.getIndex().get_id())); } } } if (failedRecords.size() > 0) { logger.warn("{} records were not written to annotation ES", failedRecords.size()); monitorService.modifyCounter(MonitorService.Counter.ANNOTATION_RECORDS_WRITE_FAILURES, failedRecords.size(), null); } } catch (IOException e) { throw new SystemException("Failed to index annotation to ES.", e); } } @Inject  ElasticSearchAnnotationService(SystemConfiguration config, MonitorService monitorService, ElasticSearchUtils esUtils); @Override void dispose(); @Override void putAnnotations(List<Annotation> annotations); @Override List<Annotation> getAnnotations(List<AnnotationQuery> queries); static final int ANNOTATION_INDEX_MAX_RESULT_WINDOW; static String ANNOTATION_INDEX_TEMPLATE_PATTERN_START; }
@Test public void testPutAnnotationsUsingAnnotationIndex() throws IOException { ElasticSearchAnnotationService spyService = _initializeSpyService(esAnnotationService, createSucessReply, true); List<Annotation> annotations = new ArrayList<>(); Annotation record1 = new Annotation("unittest", "id456", "unittest", "scope1", "metric1", 1557800720441L); annotations.add(record1); spyService.putAnnotations(annotations); ArgumentCaptor<String> requestCaptorUrl = ArgumentCaptor.forClass(String.class); ArgumentCaptor<String> requestCaptorBody = ArgumentCaptor.forClass(String.class); verifyStatic(ElasticSearchUtils.class, times(1)); ElasticSearchUtils.performESRequest(eq(restClient), requestCaptorUrl.capture(), requestCaptorBody.capture()); String expectedURL = "_bulk"; assertEquals(expectedURL, requestCaptorUrl.getValue()); String[] lines = requestCaptorBody.getValue().split("\\r?\\n"); String expectedIndexName = "argus-annotation-2019-05"; JsonNode root = mapper.readTree(lines[0]); String actualIndexName = root.get("index").get("_index").asText(); assertEquals(expectedIndexName, actualIndexName); String expectedSerializedAnnotation = "{\"metric\":\"metric1\",\"scope\":\"scope1\",\"source\":\"unittest\",\"id\":\"id456\",\"type\":\"unittest\",\"fields\":\"{}\",\"tags\":\"{}\",\"sid\":\"f9c22bcbd813474ec99f7011ae50b080\",\"ts\":\"1557800720441\"}"; assertEquals(expectedSerializedAnnotation, lines[1]); }