ElasticSearchQueryStoreService extends DefaultService implements QueryStoreService { protected Set<QueryStoreRecord> upsertQueryStoreRecords(Set<QueryStoreRecord> records) { String requestUrl = String.format("/%s/%s/_bulk", QUERY_STORE_INDEX_NAME, QUERY_STORE_TYPE_NAME); String strResponse; ObjectMapper localObjectMapper = new ObjectMapper(); QueryStoreRecordList indexQueryStoreRecordList = new QueryStoreRecordList(records, idgenHashAlgo); try { String requestBody = queryStoreMapper.writeValueAsString(indexQueryStoreRecordList); Request request = new Request(HttpMethod.POST.getName(), requestUrl); request.setEntity(new StringEntity(requestBody, ContentType.APPLICATION_JSON)); Response response = esRestClient.performRequest(request); strResponse = extractResponse(response); } catch (IOException e) { throw new SystemException("Failed to insert query store record to ES. ", e); } try { Set<QueryStoreRecord> failedRecords = new HashSet<>(); PutResponse putResponse = localObjectMapper.readValue(strResponse, PutResponse.class); if(putResponse.errors) { for(PutResponse.Item item : putResponse.items) { if (item.index !=null && item.index.status != HttpStatus.SC_CREATED) { logger.warn("Failed to add record {} to index {}. Reason: {}", indexQueryStoreRecordList.getRecord(item.index._id), QUERY_STORE_INDEX_NAME, localObjectMapper.writeValueAsString(item.index.error)); failedRecords.add(indexQueryStoreRecordList.getRecord(item.index._id)); } } if (failedRecords.size() != 0) { logger.warn("{} records were not written to index {}", failedRecords.size(), QUERY_STORE_INDEX_NAME); } } return failedRecords; } catch (IOException e) { throw new SystemException("Failed to parse response of querystore ES records insertion. The response was: " + strResponse, e); } } @Inject  ElasticSearchQueryStoreService(SystemConfiguration config, MonitorService monitorService); @Override void putArgusWsQueries(List<Metric> metrics); @Override void dispose(); @Override Properties getServiceProperties();  }
@Test public void testPutIndexUsingQueryStoreIndex() throws IOException { List<Metric> metrics = new ArrayList<>(); Metric myMetric = new Metric("scope1", "metric1"); metrics.add(myMetric); ElasticSearchQueryStoreService spyService = _initializeSpyService(_esQueryStoreService, createSucessReply, createSucessReply); Set<QueryStoreRecord> records = new HashSet<>(); for(Metric m : metrics) { QueryStoreRecord qsr = new QueryStoreRecord(m.getScope(),m.getMetric()); records.add(qsr); } spyService.upsertQueryStoreRecords(records); ArgumentCaptor<Request> requestCaptor = ArgumentCaptor.forClass(Request.class); verify(restClient, times(1)).performRequest(requestCaptor.capture()); Request capturedRequest = requestCaptor.getValue(); String createJson = EntityUtils.toString(capturedRequest.getEntity()); assertTrue(createJson.contains("index")); assertTrue(createJson.contains("mts")); assertTrue(createJson.contains("sourcehost")); assertTrue(createJson.contains("scope")); assertTrue(createJson.contains("metric")); String expectedURL = String.format("/%s/%s/_bulk", queryStoreIndexName, queryStoreTypeName); assertEquals(expectedURL, capturedRequest.getEndpoint()); }