This is an automated email from the ASF dual-hosted git repository. ningjiang pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git
commit a888641e1fafd6e1a89742dd2e2e7f85bd5b1670 Author: Lei Zhang <[email protected]> AuthorDate: Fri Jul 26 18:41:51 2019 +0800 SCB-1369 Add paging query event list method and single event query method for APIs --- .../fsm/repository/NoneTransactionRepository.java | 11 +++ .../fsm/repository/TransactionRepository.java | 7 ++ .../ElasticsearchTransactionRepository.java | 99 ++++++++++++++++++++-- .../pack/alpha/server/api/APIControllerV1.java | 24 ++++++ 4 files changed, 135 insertions(+), 6 deletions(-) diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/repository/NoneTransactionRepository.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/repository/NoneTransactionRepository.java index c0cc566..9cd6d1e 100644 --- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/repository/NoneTransactionRepository.java +++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/repository/NoneTransactionRepository.java @@ -19,6 +19,7 @@ package org.apache.servicecomb.pack.alpha.fsm.repository; import java.lang.invoke.MethodHandles; import org.apache.servicecomb.pack.alpha.fsm.repository.model.GloablTransaction; +import org.apache.servicecomb.pack.alpha.fsm.repository.model.PagingGloablTransactions; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -30,4 +31,14 @@ public class NoneTransactionRepository implements TransactionRepository { public void send(GloablTransaction transaction) { LOG.warn("Please configure alpha.feature.akka.transaction.repository.type=elasticsearch to use elasticsearch to persist transaction data."); } + + @Override + public GloablTransaction getGloablTransactionByGlobalTxId(String globalTxId){ + throw new UnsupportedOperationException("NoneTransactionRepository Unsupported!"); + } + + @Override + public PagingGloablTransactions getGloablTransactions(int page, int size) throws Exception { + throw new UnsupportedOperationException("NoneTransactionRepository Unsupported!"); + } } diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/repository/TransactionRepository.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/repository/TransactionRepository.java index 1916354..472aabc 100644 --- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/repository/TransactionRepository.java +++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/repository/TransactionRepository.java @@ -18,8 +18,15 @@ package org.apache.servicecomb.pack.alpha.fsm.repository; import org.apache.servicecomb.pack.alpha.fsm.repository.model.GloablTransaction; +import org.apache.servicecomb.pack.alpha.fsm.repository.model.PagingGloablTransactions; public interface TransactionRepository { void send(GloablTransaction transaction) throws Exception; + + GloablTransaction getGloablTransactionByGlobalTxId(String globalTxId) + throws Exception; + + PagingGloablTransactions getGloablTransactions(int page, int size) + throws Exception; } diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/repository/elasticsearch/ElasticsearchTransactionRepository.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/repository/elasticsearch/ElasticsearchTransactionRepository.java index b7dc0a1..ef62b18 100644 --- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/repository/elasticsearch/ElasticsearchTransactionRepository.java +++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/repository/elasticsearch/ElasticsearchTransactionRepository.java @@ -19,22 +19,38 @@ package org.apache.servicecomb.pack.alpha.fsm.repository.elasticsearch; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.IOException; import java.lang.invoke.MethodHandles; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import org.apache.servicecomb.pack.alpha.fsm.metrics.MetricsService; import org.apache.servicecomb.pack.alpha.fsm.repository.TransactionRepository; import org.apache.servicecomb.pack.alpha.fsm.repository.model.GloablTransaction; +import org.apache.servicecomb.pack.alpha.fsm.repository.model.PagingGloablTransactions; +import org.elasticsearch.action.search.SearchResponse; +import org.elasticsearch.index.query.QueryBuilders; +import org.elasticsearch.search.SearchHit; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.data.domain.PageRequest; +import org.springframework.data.domain.Pageable; import org.springframework.data.elasticsearch.core.ElasticsearchTemplate; +import org.springframework.data.elasticsearch.core.ScrolledPage; +import org.springframework.data.elasticsearch.core.SearchResultMapper; +import org.springframework.data.elasticsearch.core.aggregation.AggregatedPage; +import org.springframework.data.elasticsearch.core.aggregation.impl.AggregatedPageImpl; +import org.springframework.data.elasticsearch.core.query.GetQuery; import org.springframework.data.elasticsearch.core.query.IndexQuery; +import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder; +import org.springframework.data.elasticsearch.core.query.SearchQuery; public class ElasticsearchTransactionRepository implements TransactionRepository { private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); - private static final String INDEX_NAME = "alpha_global_transaction"; - private static final String INDEX_TYPE = "alpha_global_transaction_type"; + public static final String INDEX_NAME = "alpha_global_transaction"; + public static final String INDEX_TYPE = "alpha_global_transaction_type"; + private static final long SCROLL_TIMEOUT = 3000; private final ElasticsearchTemplate template; private final MetricsService metricsService; private final ObjectMapper mapper = new ObjectMapper(); @@ -54,7 +70,7 @@ public class ElasticsearchTransactionRepository implements TransactionRepository if (this.refreshTime > 0) { new Thread(new RefreshTimer(), "elasticsearch-repository-refresh").start(); } - if(!this.template.indexExists(INDEX_NAME)){ + if (!this.template.indexExists(INDEX_NAME)) { this.template.createIndex(INDEX_NAME); } } @@ -66,7 +82,7 @@ public class ElasticsearchTransactionRepository implements TransactionRepository batchSizeCounter++; metricsService.metrics().doRepositoryReceived(); if (batchSize == 0 || batchSizeCounter == batchSize) { - synchronized (lock){ + synchronized (lock) { save(begin); batchSizeCounter = 0; queries.clear(); @@ -74,6 +90,77 @@ public class ElasticsearchTransactionRepository implements TransactionRepository } } + @Override + public GloablTransaction getGloablTransactionByGlobalTxId(String globalTxId) { + GetQuery getQuery = new GetQuery(); + getQuery.setId(globalTxId); + GloablTransactionDocument gloablTransaction = this.template + .queryForObject(getQuery, GloablTransactionDocument.class); + return gloablTransaction; + } + + @Override + public PagingGloablTransactions getGloablTransactions(int page, int size) { + long start = System.currentTimeMillis(); + List<GloablTransaction> gloablTransactions = new ArrayList(); + SearchQuery searchQuery = new NativeSearchQueryBuilder() + .withIndices(INDEX_NAME) + .withTypes(INDEX_TYPE) + .withQuery(QueryBuilders.matchAllQuery()) + .withPageable(PageRequest.of(page, size)) + .build(); + + ScrolledPage<GloablTransactionDocument> scroll = (ScrolledPage<GloablTransactionDocument>) this.template + .startScroll(SCROLL_TIMEOUT, searchQuery, GloablTransactionDocument.class, + searchResultMapper); + int pageCursor = 0; + while (scroll.hasContent()) { + if (pageCursor < page) { + scroll = (ScrolledPage<GloablTransactionDocument>) this.template + .continueScroll(scroll.getScrollId(), SCROLL_TIMEOUT, GloablTransactionDocument.class, + searchResultMapper); + pageCursor++; + } else { + for (GloablTransactionDocument dto : scroll.getContent()) { + gloablTransactions.add(dto); + } + break; + } + } + LOG.info("Query total hits {}, return page {}, size {}", scroll.getTotalElements(), page, size); + this.template.clearScroll(scroll.getScrollId()); + return PagingGloablTransactions.builder().page(page).size(size).total(scroll.getTotalElements()) + .gloablTransactions(gloablTransactions).elapsed(System.currentTimeMillis() - start).build(); + } + + private final SearchResultMapper searchResultMapper = new SearchResultMapper() { + @Override + public <T> AggregatedPage<T> mapResults(SearchResponse response, Class<T> aClass, + Pageable pageable) { + List<GloablTransactionDocument> result = new ArrayList<>(); + for (SearchHit hit : response.getHits()) { + if (response.getHits().getHits().length <= 0) { + return new AggregatedPageImpl<T>(Collections.EMPTY_LIST, pageable, + response.getHits().getTotalHits(), response.getScrollId()); + } + GloablTransactionDocument gloablTransactionDocument = null; + try { + gloablTransactionDocument = mapper.readValue(hit.getSourceAsString(), + GloablTransactionDocument.class); + } catch (IOException e) { + new RuntimeException(e.getMessage(), e); + } + result.add(gloablTransactionDocument); + } + if (result.isEmpty()) { + return new AggregatedPageImpl<T>(Collections.EMPTY_LIST, pageable, + response.getHits().getTotalHits(), response.getScrollId()); + } + return new AggregatedPageImpl<T>((List<T>) result, pageable, + response.getHits().getTotalHits(), response.getScrollId()); + } + }; + private IndexQuery convert(GloablTransaction transaction) throws JsonProcessingException { IndexQuery indexQuery = new IndexQuery(); indexQuery.setId(transaction.getGlobalTxId()); @@ -83,7 +170,7 @@ public class ElasticsearchTransactionRepository implements TransactionRepository return indexQuery; } - private void save(long begin){ + private void save(long begin) { template.bulkIndex(queries); template.refresh(INDEX_NAME); metricsService.metrics().doRepositoryAccepted(queries.size()); @@ -97,7 +184,7 @@ public class ElasticsearchTransactionRepository implements TransactionRepository public void run() { while (true) { try { - synchronized (lock){ + synchronized (lock) { if (!queries.isEmpty()) { save(System.currentTimeMillis()); } diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/api/APIControllerV1.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/api/APIControllerV1.java index 482acf6..0c0481e 100644 --- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/api/APIControllerV1.java +++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/api/APIControllerV1.java @@ -17,11 +17,16 @@ package org.apache.servicecomb.pack.alpha.server.api; +import org.apache.servicecomb.pack.alpha.fsm.repository.TransactionRepository; +import org.apache.servicecomb.pack.alpha.fsm.repository.model.GloablTransaction; +import org.apache.servicecomb.pack.alpha.fsm.repository.model.PagingGloablTransactions; import org.apache.servicecomb.pack.alpha.server.metrics.AlphaMetrics; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; @RestController @@ -31,9 +36,28 @@ public class APIControllerV1 { @Autowired AlphaMetrics AlphaMetrics; + @Autowired + TransactionRepository transactionRepository; + @GetMapping(value = "/metrics") ResponseEntity<AlphaMetrics> metrics() { return ResponseEntity.ok(AlphaMetrics); } + @GetMapping(value = "/transaction/{globalTxId}") + ResponseEntity<GloablTransaction> getTransactionByGlobalTxId(@PathVariable String globalTxId) + throws Exception { + GloablTransaction gloablTransaction = transactionRepository + .getGloablTransactionByGlobalTxId(globalTxId); + return ResponseEntity.ok(gloablTransaction); + } + + @GetMapping(value = "/transaction") + ResponseEntity<PagingGloablTransactions> getTransactions(@RequestParam(value = "page", required = false, defaultValue = "0") int page, + @RequestParam(value = "size", required = false, defaultValue = "50") int size) + throws Exception { + PagingGloablTransactions pagingGloablTransactions = transactionRepository + .getGloablTransactions(page, size); + return ResponseEntity.ok(pagingGloablTransactions); + } }
